Select

下面给你一份 SQLAlchemy 2.0 ORM select() 最清晰、最完整、最实战的教程，涵盖基础、进阶、JOIN、过滤、多表、分页、聚合、子查询、返回值处理等内容。

内容经过整理，完全适配 SQLAlchemy 2.0（sync + async 通用）。

1. 基础：最简单的 select

from sqlalchemy import select
from models import User

stmt = select(User)
result = session.execute(stmt).scalars().all()

返回格式（ORM 模式）

result.scalars().all()  ->  list[User]

2. 只查询部分字段

stmt = select(User.id, User.name)
rows = session.execute(stmt).all()

返回格式

[(1, "martin"), (2, "john"), ...]

3. 使用过滤条件（=、>、<、LIKE、IN）

3.1 等于

stmt = select(User).where(User.status == 1)

3.2 多条件 AND

stmt = select(User).where(
    User.age > 18,
    User.status == 1
)

3.3 OR 条件

from sqlalchemy import or_

stmt = select(User).where(
    or_(User.age < 18, User.status == 0)
)

3.4 IN

stmt = select(User).where(User.id.in_([1, 2, 3]))

3.5 LIKE

stmt = select(User).where(User.name.like("%martin%"))

4. 排序 ORDER BY

stmt = select(User).order_by(User.id.desc())

5. 分页 LIMIT OFFSET

stmt = (
    select(User)
    .order_by(User.id.desc())
    .limit(10)
    .offset(0)
)

6. JOIN（最常用）

内连接 JOIN

stmt = (
    select(User, Profile)
    .join(Profile, Profile.user_id == User.id)
)

返回值示例：

[(User, Profile), (User, Profile), ...]

LEFT JOIN

stmt = (
    select(User, Profile)
    .outerjoin(Profile, Profile.user_id == User.id)
)

7. 多表 JOIN + 指定字段

stmt = (
    select(User.id, User.name, Profile.age)
    .join(Profile, Profile.user_id == User.id)
)

返回格式：

[(1, "Martin", 25), ...]

8. DISTINCT（去重）

stmt = select(User.name).distinct()

9. GROUP BY、聚合函数

from sqlalchemy import func

stmt = (
    select(User.status, func.count())
    .group_by(User.status)
)

[(1, 10), (0, 3)]

10. 子查询 Subquery

子查询构建

subq = (
    select(User.id)
    .where(User.status == 1)
).subquery()

主查询使用

stmt = select(Profile).where(Profile.user_id.in_(subq))

11. exists / not exists

from sqlalchemy import exists

subq = select(User.id).where(User.id == Profile.user_id)

stmt = select(Profile).where(exists(subq))

12. scalar()、scalars()、mappings() 返回值解释

scalar()

返回单个值，例如 COUNT、MAX：

count = session.execute(select(func.count()).select_from(User)).scalar()

scalars()

返回 ORM 对象或单个列：

names = session.execute(select(User.name)).scalars().all()

mappings()

返回 dict-like 结构：

rows = session.execute(select(User.id, User.name)).mappings().all()

返回示例：

[{'id': 1, 'name': 'martin'}]

13. 多条件动态构建查询（实战最常用）

filters = []

if status is not None:
    filters.append(User.status == status)

if keyword:
    filters.append(User.name.like(f"%{keyword}%"))

stmt = select(User).where(*filters)

14. ORM 模型与 SQL 拼接输出 SQL

print(stmt)

或者：

print(stmt.compile(compile_kwargs={"literal_binds": True}))

15. AsyncSession（异步版本）

同步只需换成 async 版本：

result = await session.execute(stmt)
rows = result.scalars().all()

完整示例：

async with async_session() as session:
    stmt = select(User).where(User.status == 1)
    result = await session.execute(stmt)
    users = result.scalars().all()

📌 完整示例：SELECT + JOIN + 分页 + 排序 + 搜索

stmt = (
    select(User.id, User.name, Profile.age)
    .join(Profile, Profile.user_id == User.id)
    .where(
        User.status == 1,
        User.name.like(f"%{keyword}%"),
    )
    .order_by(User.id.desc())
    .limit(10)
    .offset(0)
)

rows = session.execute(stmt).mappings().all()