SQLAlchemy 2.x 異步查詢中常用的 結果處理方法速查表,包含方法說明、使用場景、返回類型及典型用途。
SQLAlchemy 查詢結果處理方法速查表(適用于 AsyncSession)
方法 | 說明 | 返回類型 | 示例 SQL | 示例輸出 |
| 獲取單列所有值 |
|
|
|
| 獲取單列的第一行 |
|
|
|
| 獲取第一行的第一列(聚合) |
|
|
|
| 多列,每行為 dict 映射 |
|
|
|
| 返回第一行 dict 映射 |
|
|
|
| 返回第一行的 Row | `Row | None` |
|
| 返回所有行 Row 對象列表 |
|
|
|
| 返回一行 Row |
|
|
|
| 返回所有 Row(非字典) |
|
|
|
| 獲取唯一一行,若多行/無行報錯 |
|
|
|
| 獲取一行或 None | `Row | None` |
|
使用建議(按類型):
- 獲取單列(如用戶 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()
替代
- 老式用法,返回
推薦使用方式總結:
查詢類型 | 推薦方式 |
單列、多行 |
|
單列、單值 |
|
多列、多行 |
|
多列、單行 |
|
真實使用示例(異步 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': '123@qq.com'},# {'id': 3, 'user_name': 'test1', 'email': '456@qq.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': '123@qq.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', '123@qq.com'), (3, 'test1', '456@qq.com')]
?