SQLAlchemy教程：第一个例子-7.实现查询 作者：马育民 • 2026-08-16 19:12 • 阅读：10000 # 介绍 执行增删改查时，要使用 `session` 上下文管理器，执行后自动关闭会话，不用手动关闭： ``` async with AsyncSession(async_engine) as session: # 根据id更新数据 stmt = select(User).where(User.id == id) # 执行 res = await session.execute(stmt) ``` ### 查询返回值 1. `res.scalar()`：获取第一行第一个字段 2. `res.scalar_one_or_none()`：查询单条ORM对象，找不到返回None 3. `res.scalars().all()`：获取全部ORM模型对象列表（ORM查询优先用） 4. `res.mappings().all()`：返回字典行，适合原生text() SQL **❌ 错误：**不要直接 `session.execute(stmt).all()`，异步必须加 await # 实现查询所有数据 ### 导包 ``` from sqlalchemy import select ``` ### 定义查询函数 ``` async def query_all(): """查询所有数据""" async with AsyncSession(async_engine) as session: stmt = select(User) res = await session.execute(stmt) user_list = res.scalars().all() # scalars获取模型对象 return user_list ``` ### 执行 ``` async def main(): await init_table() # 查询所有数据 user_list = await query_all() print("查询结果：开始------------") for u in user_list: print(f"id={u.id}, username={u.username}, password={u.password}") print("查询结果：结束------------") # ✅关键：主动dispose释放全部连接，避免GC阶段触发报错 await async_engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` # 实现根据id查询数据 ### 定义查询函数 ``` async def query_by_id(id:int): """查询一条数据""" async with AsyncSession(async_engine) as session: stmt = select(User).where(User.id == id) res = await session.execute(stmt) # 返回一条数据，或没有数据 user = res.scalar_one_or_none() return user ``` ### 执行 ``` async def main(): await init_table() # 查询所有数据 # await query_all() user = await query_by_id(3) print(f"查询结果：id={user.id},username={user.username},password={user.password}") # ✅关键：主动dispose释放全部连接，避免GC阶段触发报错 await async_engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` # 实现查询原生SQL ### 导包 ``` from sqlalchemy import text ``` ### 定义查询函数 ``` async def query_sql(sql:str,params:tuple): """查询原生SQL""" async with AsyncSession(async_engine) as session: sql_t = text(sql) result = await session.execute(sql_t, params) rows = result.mappings().all() return rows ``` ### 执行 ``` async def main(): await init_table() # 查询原生SQL res = await query_sql("SELECT * FROM user where username = :username", {"username": 'lilei'}) print("查询结果：开始------------") for u in res: print(f"id={u.id}, username={u.username}, password={u.password}") print("查询结果：结束------------") # ✅关键：主动dispose释放全部连接，避免GC阶段触发报错 await async_engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` # 完整代码 ``` import asyncio from sqlalchemy import String, Integer, insert, delete, update, select, text 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 query_all(): """查询所有数据""" async with AsyncSession(async_engine) as session: stmt = select(User) res = await session.execute(stmt) user_list = res.scalars().all() # scalars获取模型对象 return user_list async def query_by_id(id:int): """查询一条数据""" async with AsyncSession(async_engine) as session: stmt = select(User).where(User.id == id) res = await session.execute(stmt) # 返回一条数据，或没有数据 user = res.scalar_one_or_none() return user async def query_sql(sql:str,params:tuple): """查询原生SQL""" async with AsyncSession(async_engine) as session: sql_t = text(sql) result = await session.execute(sql_t, params) rows = result.mappings().all() return rows async def main(): await init_table() # 查询所有数据 # user_list = await query_all() # # print("查询结果：开始------------") # for u in user_list: # print(f"id={u.id}, username={u.username}, password={u.password}") # # print("查询结果：结束------------") # user = await query_by_id(3) # print(f"查询结果：id={user.id},username={user.username},password={user.password}") # 查询原生SQL res = await query_sql("SELECT * FROM user where username = :username", {"username": 'lilei'}) print("查询结果：开始------------") for u in res: print(f"id={u.id}, username={u.username}, password={u.password}") print("查询结果：结束------------") # ✅关键：主动dispose释放全部连接，避免GC阶段触发报错 await async_engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` 原文出处：/show_1GW3sEgNaGyf.html