蚌埠做网站哪家好,西安 医疗网站建设,电子商务网站有哪些和网址,营销型网站试运营调忧SQLAlchemy 2.x 异步查询中常用的 结果处理方法速查表#xff0c;包含方法说明、使用场景、返回类型及典型用途。
SQLAlchemy 查询结果处理方法速查表#xff08;适用于 AsyncSession#xff09; 方法 说明 返回类型 示例 SQL 示例输出 scalars().all() 获取单列所有…SQLAlchemy 2.x 异步查询中常用的 结果处理方法速查表包含方法说明、使用场景、返回类型及典型用途。
SQLAlchemy 查询结果处理方法速查表适用于 AsyncSession 方法 说明 返回类型 示例 SQL 示例输出 scalars().all() 获取单列所有值 List[Any] select(User.id) [1, 2, 3, 4] scalars().first() 获取单列的第一行 Any select(User.email) aliceexample.com scalar() 获取第一行的第一列聚合 Any select(func.count(User.id)) 27 mappings().all() 多列每行为 dict 映射 List[dict] select(User.id, User.email) [{id: 1, email: axx.com}, {id: 2, email: bxx.com}] mappings().first() 返回第一行 dict 映射 dict select(User.id, User.name) {id: 1, name: Alice} first() 返回第一行的 Row Row None select(User.id, User.name) fetchall() 返回所有行 Row 对象列表 List[Row] select(User.id, User.name) [(1, Alice), (2, Bob)] fetchone() 返回一行 Row Row select(User.id, User.name) (1, Alice) all() 返回所有 Row非字典 List[Row] select(User.id, User.name) [(1, Alice), (2, Bob)] one() 获取唯一一行若多行/无行报错 Row select(User).where(User.id1) User(id1, nameAlice) one_or_none() 获取一行或 None Row None select(User).where(User.id9999)
使用建议按类型
获取单列如用户 ID → 用 scalars().all()统计总数/最大值 → 用 scalar()获取结构化多列数据用于 JSON 返回 → 用 mappings().all()查找某条记录详情 → 用 mappings().first() 或 one_or_none()返回原始 SQL 风格 Row → 用 fetchall() / all()不推荐
详细解释与建议
scalars() 专门用于查询 单列如 select(User.id)会自动去除 Row 封装只返回具体值
mappings() 用于查询 多列将每一行封装为 dict字段名值最常用在结构化响应返回中如 JSON
first()、one() first()最多一行超过无所谓返回第一行one()必须返回一行返回多行或无行都抛异常
fetchall() / fetchone() 老式用法返回 Row 对象类似元组不推荐在现代异步代码中使用建议用 scalars() / mappings() 替代
推荐使用方式总结 查询类型 推荐方式 单列、多行 scalars().all() 单列、单值 scalar()或 scalars().first() 多列、多行 mappings().all() 多列、单行 mappings().first()或 one_or_none()
真实使用示例异步 FastAPI
async def test_get_single_column():async with session_factory() as session:query select(User.id)result await session.execute(query)user_ids result.scalars().all()print(用户 ID 列表:, user_ids) # 用户 ID 列表: [1, 3]# 2. 统计总数/最大值 → 用 scalar()
async def test_aggregate_functions():async with session_factory() as session:# 统计用户总数total_count await session.execute(select(func.count(User.id)))total total_count.scalar()print(用户总数:, total) # 用户总数: 2# 查找最大的角色 IDmax_role_id await session.execute(select(func.max(User.role_id)))max_role_id_value max_role_id.scalar()print(最大的角色 ID:, max_role_id_value) # 最大的角色 ID: 2# 3. 获取结构化多列数据用于 JSON 返回 → 用 mappings().all()
async def test_get_structured_data():async with session_factory() as session:query select(User.id, User.user_name, User.email)result await session.execute(query)users result.mappings().all()print(结构化多列数据:, users)# 结构化多列数据: [{id: 1, user_name: test, email: 123qq.com},# {id: 3, user_name: test1, email: 456qq.com}]# 4. 查找某条记录详情 → 用 mappings().first() 或 one_or_none()
async def test_find_single_record():async with session_factory() as session:# 使用 mappings().first()query select(User.id, User.user_name, User.email).where(User.id 1)result await session.execute(query)user result.mappings().first()print(使用 mappings().first() 查找的记录:, user)# 使用 mappings().first() 查找的记录: {id: 1, user_name: test, email: 123qq.com}# 使用 one_or_none()user_obj await session.execute(select(User).where(User.id 1))user_result user_obj.scalars().one_or_none()print(使用 one_or_none() 查找的记录:, user_result)# 使用 one_or_none() 查找的记录: login_related.model.user.User object at 0x000001477DC206D0# 5. 返回原始 SQL 风格 Row → 用 fetchall() / all()不推荐
async def test_get_raw_rows():async with session_factory() as session:query select(User.id, User.user_name, User.email)result await session.execute(query)rows result.all()print(原始 SQL 风格 Row:, rows)# 原始 SQL 风格 Row: [(1, test, 123qq.com), (3, test1, 456qq.com)]
