1. 从表到对象

ORM 的核心作用，是在数据库结构与 Python 对象之间建立一组明确映射。下面的示例沿用 Task、Project 等模型，并假设调用方已经取得一个 AsyncSession。阅读代码前，可以先记住这些对应关系：

这种映射并没有取消 SQL。select(Task) 最终仍会生成 SELECT，修改对象后 flush 仍会生成 UPDATE，数据库约束也仍然由 PostgreSQL 执行。ORM 解决的是对象组织、状态跟踪和关系加载问题，而不是用 Python 规则替代数据库。

Session 会先记录本次工作中新增、修改和删除了哪些对象，再在 flush 时按依赖顺序生成 SQL。这种把一组对象变化统一同步到数据库的机制称为 Unit of Work（工作单元）。它解释了为什么修改已加载对象时不用立即写 UPDATE，也解释了为什么一次 flush 可能同时发送多条 SQL。

刚执行 Task(title="学习 ORM", ...) 时，只创建了一个普通 Python 对象，数据库中还没有对应记录。对象加入 Session 并 flush 后，SQLAlchemy 才会发送 INSERT；从数据库查询得到的 Task，则是由查询结果构造并纳入当前 Session 管理的对象。

2. 查询

ORM 查询使用 select() 构造语句，再交给 Session 执行。下面这段代码对应“查询项目中未归档的任务，并按创建时间倒序取前 20 条”：

01
from sqlalchemy import select
02

03
from app.db.models.task import Task
04

05

06
statement = (
07
    select(Task)
08
    .where(
09
        Task.project_id == project_id,
10
        Task.archived_at.is_(None),
11
    )
12
    .order_by(Task.created_at.desc(), Task.id.desc())
13
    .limit(20)
14
)
15

16
result = await session.scalars(statement)
17
tasks = result.all()

select(Task) 表示选择完整的 Task 实体，多个 .where() 条件默认通过 AND 连接。Task.archived_at.is_(None) 会生成 IS NULL，不能写成 Python 的 is None，因为后者只会立即得到一个布尔值，无法构造 SQL 条件。

这段表达式生成的 SQL 大致如下。具体列清单和参数占位符会由方言与驱动决定：

1
SELECT tasks.*
2
FROM tasks
3
WHERE tasks.project_id = :project_id
4
  AND tasks.archived_at IS NULL
5
ORDER BY tasks.created_at DESC, tasks.id DESC
6
LIMIT :limit;

project_id 和 limit 会作为绑定参数发送，不是通过字符串拼接进入 SQL。session.scalars() 执行语句，并把每一行中的第一个实体或值提取出来；all() 再将结果收集为列表。因此，这里的 tasks 是 Task 对象列表，不是字典列表。

如果查询的是多个独立列，就应使用 execute() 保留完整的行结构：

01
statement = select(Task.id, Task.title).where(
02
    Task.project_id == project_id,
03
    Task.archived_at.is_(None),
04
)
05

06
result = await session.execute(statement)
07
rows = result.all()
08

09
for task_id, title in rows:
10
    print(task_id, title)

scalars() 适合只选择一个实体或一列的查询；同时选择多列时使用 execute()，可以避免其余列被丢弃。

按主键读取可以使用：

1
task = await session.get(Task, task_id)

get() 只用于按主键读取。它会先查看 Session 的 identity map，当前 Session 已经加载过该对象时可能不再发送 SQL；但它不会自动附加项目范围、归档状态或权限条件。需要这些限制时，应使用完整的 select()：

1
statement = select(Task).where(
2
    Task.id == task_id,
3
    Task.project_id == project_id,
4
    Task.archived_at.is_(None),
5
)
6
result = await session.execute(statement)
7
task = result.scalar_one_or_none()

这里的业务约束是“至多找到一个任务”，所以使用 scalar_one_or_none()：没有结果时返回 None，出现多行时抛出异常。要求恰好一行时使用 scalar_one()。相比之下，session.scalar() 只返回第一行的第一个值，无法发现意外的重复结果。结果方法应当体现查询预期，而不只是为了方便取值。

3. 新增

新增对象仍然通过 Session：

01
from sqlalchemy.ext.asyncio import AsyncSession
02

03
from app.db.models.task import Task
04

05

06
async def create_task(
07
    session: AsyncSession,
08
    project_id: int,
09
    task_number: int,
10
    title: str,
11
) -> Task:
12
    task = Task(
13
        project_id=project_id,
14
        task_number=task_number,
15
        title=title,
16
    )
17
    session.add(task)
18
    await session.flush()
19
    return task

session.add() 只改变 Session 中的对象状态，不执行网络 I/O，所以不需要 await。这个函数只 flush，不 commit。执行 flush 后，PostgreSQL 已经生成 task.id，服务端默认值也可以通过 RETURNING 回填，约束错误还可能在这里暴露；不过，这些变化仍然属于当前事务，可以回滚。调用它的服务可以继续写入状态历史或其他数据，最后统一提交。

不要把 Pydantic 请求对象直接展开为所有 ORM 字段。owner_id、created_at、权限字段等应由服务端上下文决定：

1
task = Task(
2
    project_id=authorized_project.id,
3
    task_number=reserved_task_number,
4
    title=request.title,
5
    description=request.description,
6
    priority=request.priority,
7
)

明确映射虽然多写几行，却能清楚看出客户端允许控制哪些数据库字段。