介绍
执行增删改查时,要使用 session 上下文管理器,执行后自动关闭会话,不用手动关闭:
async with async_session() 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 async_session() 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, update
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 update_user(id:int,password:str):
"""根据id更新数据"""
async with async_session() 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(4,'777')
# ✅关键:主动dispose释放全部连接,避免GC阶段触发报错
await async_engine.dispose()
if __name__ == "__main__":
asyncio.run(main())