SQLAlchemy教程：第一个例子-7.实现查询

作者：马育民 • • 阅读：10017

介绍

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

async with async_session() 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 async_session() 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 async_session() 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

查询原生SQL，适合查询 复杂SQL，否则不推荐，因为 不能跨数据库

导包

from sqlalchemy import text

定义查询函数

async def query_sql(sql:str,params:tuple):
    """查询原生SQL，适合查询复杂SQL，否则不推荐，因为不能跨数据库"""
    async with async_session() 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, 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 query_all():
    """查询所有数据"""
    async with async_session() 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 async_session() 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，适合查询复杂SQL，否则不推荐，因为不能跨数据库"""
    async with async_session() 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

吉ICP备15003043号