介绍

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

from sqlalchemy import delete


async with async_session() as session:
    # 根据id删除数据
    stmt = delete(User).where(User.id == id)
    # 执行
    await session.execute(stmt)
    # 提交事务，不写此句不执行SQL语句
    await session.commit()

实现

导包

from sqlalchemy import delete

定义删除函数

根据id删除数据

async def delete_user(id:int):
    """根据id删除数据"""
    async with async_session() as session:
        try:
            stmt = delete(User).where(User.id == id)
            await session.execute(stmt)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e

执行

async def main():
    await init_table()

    # 根据id删除数据
    await delete_id(2)

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


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

完整代码

import asyncio

from sqlalchemy import String, Integer, delete
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 delete_user(id:int):
    """根据id删除数据"""
    async with AsyncSession(async_engine) as session:
        try:
            stmt = delete(User).where(User.id == id)
            await session.execute(stmt)
            # 成功执行才提交事务
            await session.commit()
        except Exception as e:
            # 如果报错就回滚
            await session.rollback()
            raise e


async def main():
    await init_table()

    # 根据id删除数据
    await delete_user(2)

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


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

原文出处：/show_1GW3sE3owSmB.html