SQLAlchemy教程：第一个例子-6.实现更新数据-马育民老师

# 介绍

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

```
async with AsyncSession(async_engine) as session:
	# 根据id更新数据
	stmt = update(User).where(User.id == id).values(password=password)
	# 执行
	await session.execute(stmt)
	# 提交事务，不写此句不执行SQL语句
	await session.commit()
```

# 实现

### 导包

```
from sqlalchemy import update
```

### 定义更新函数

根据id更新数据

```
async def update_user(id:int,password:str):
    """根据id更新数据"""
    async with AsyncSession(async_engine) as session:
        try:
            stmt = update(User).where(User.id == id).values(password=password)
            await session.execute(stmt)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e
```

### 执行

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

# 根据id删除数据
    await update_user(3,'777')

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

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

# 完整代码

```
import asyncio

from sqlalchemy import String, Integer, insert, delete, update
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
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,    # 最大连接数量
)

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 update_user(id:int,password:str):
    """根据id更新数据"""
    async with AsyncSession(async_engine) as session:
        try:
            stmt = update(User).where(User.id == id).values(password=password)
            await session.execute(stmt)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e

async def main():
    await init_table()

# 根据id删除数据
    await update_user(3,'777')

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

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

原文出处：/show_1GW3sEEgXgqq.html