SQLAlchemy教程：第一个例子-4.实现保存数据-马育民老师

# 介绍

执行增删改查时，要使用 `session` 上下文管理器，执行后自动关闭会话，不用手动关闭：

```
async with async_session() as session:
	# 保存表模型对象
	session.add(user)
	# 提交事务，不写此句不执行SQL语句
	await session.commit()
```

# 实现

**提示：**本例使用工厂写法

### 定义保存函数

```
async def save(user: User):
    # 使用上下文管理器，自动关闭会话
    async with async_session() as session:
        try:
            session.add(user)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e
```

### 执行

```
async def main():
    await init_table()

# 创建对象，传入数据
    user = User(username="lilei", password="123456")
    # 调用save方法
    await save(user)

# ✅关键：主动dispose释放全部连接，避免GC阶段触发报错
    await async_engine.dispose()

if __name__ == "__main__":
    asyncio.run(main())
```

# 完整代码

```
import asyncio

from sqlalchemy import String, Integer
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

# 数据库链接
DB_URL = "mysql+aiomysql://root:root@127.0.0.1:3306/fastapi_std?charset=utf8mb4"

async_engine = create_async_engine(
    DB_URL,
    echo=True,          # True打印执行SQL，调试打开；生产关闭
    pool_size=5,        # 连接池大小
    max_overflow=10,    # 最大连接数量
)

"""
创建异步会话工厂，通过该会话工厂，创建session
好处：参数统一在这里配置，如：expire_on_commit
"""
async_session = async_sessionmaker(
    async_engine,           # 绑定数据库引擎
    class_=AsyncSession,    # 指定会话类，调用 async_session() 返回的类型
    expire_on_commit=False  # commit后，会话中的模型对象不过期，访问对象的属性，不需重新查询数据库
)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user"
    id: Mapped[int] = mapped_column(Integer(),primary_key=True,autoincrement=True,comment="主键")
    username: Mapped[str] = mapped_column(String(50),comment="")
    password: Mapped[int] = mapped_column(String(50),comment="密码")

async def init_table():
    async with async_engine.begin() as conn:
        # 删除旧表（测试用）
        # await conn.run_sync(Base.metadata.drop_all)
        # 创建表
        await conn.run_sync(Base.metadata.create_all)

async def save(user: User):
    # 使用上下文管理器，自动关闭会话
    async with async_session() as session:
        try:
            session.add(user)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e

async def main():
    await init_table()

# 添加数据
    user = User(username="lilei", password="123456")
    await save(user)
    print("id:", user.id)

# ✅关键：主动dispose释放全部连接，避免GC阶段触发报错
    await async_engine.dispose()

if __name__ == "__main__":
    asyncio.run(main())
```

原文出处：/show_1GW3sXARZvIQ.html