API、Service、Repository：三层怎么拆

一个文件里从 request.json() 写到 session.execute(text("UPDATE ..."))，小 demo 能跑；功能一多，路由函数变成几百行，改个字段要 grep 全项目，测试只能 TestClient 端到端硬扫。分层不是教条，而是把变化速度不同的代码隔开：HTTP 形状变得快，业务规则中等，SQL 与表结构最慢。常见拆法是 API（路由）→ Service（业务）→ Repository（数据访问），依赖单向向下，上层不知道下层用的是 PostgreSQL 还是内存 dict。

三层各自干什么 ​

API 层只做「边界翻译」：Pydantic schema 进，调 OrderService.create(...)，把 AppError 或返回值变成 response model。Service 层回答「这样业务上对不对」：库存够不够、优惠券能否叠加。Repository 层回答「怎么持久化」：get_by_id、list_by_tenant，隐藏 SQLAlchemy 细节。

变化频率不同是分层的主因：前端改表单字段 → 动 schemas；促销规则改 → 动 Service；加索引改查询 → 动 Repository。混在一起时，改一个校验会误触 SQL，或改索引破坏业务测试。

依赖方向 ​

  HTTP Request
       ↓
   API (router)
       ↓
   Service
       ↓
   Repository
       ↓
     Database

Service 依赖 Repository 接口（抽象类或 Protocol），不依赖具体 SqlAlchemyOrderRepository 的类型在路由里——路由只依赖 Service。测试 Service 时注入 InMemoryOrderRepository 即可。

禁止 Repository 调 Service，禁止 Service import FastAPI 的 Request。若 Repository 需要当前租户 ID，由 Service 把 tenant_id 当参数传入查询方法，而不是 Repository 里 Depends(get_tenant)——数据层应保持框架无关。

代码骨架示例 ​

领域与持久化：

from dataclasses import dataclass
from typing import Protocol


@dataclass
class Order:
    id: int | None
    tenant_id: str
    total: float


class OrderRepository(Protocol):
    def get(self, order_id: int, tenant_id: str) -> Order | None: ...
    def save(self, order: Order) -> Order: ...


class SqlAlchemyOrderRepository:
    def __init__(self, session: Session) -> None:
        self.session = session

    def get(self, order_id: int, tenant_id: str) -> Order | None:
        row = (
            self.session.query(OrderModel)
            .filter(OrderModel.id == order_id, OrderModel.tenant_id == tenant_id)
            .first()
        )
        return row.to_domain() if row else None

    def save(self, order: Order) -> Order:
        ...

Service：

class OrderService:
    def __init__(self, repo: OrderRepository) -> None:
        self.repo = repo

    def get_order(self, order_id: int, tenant_id: str) -> Order:
        order = self.repo.get(order_id, tenant_id)
        if order is None:
            raise AppError(code="ORDER_NOT_FOUND", message="order not found")
        return order

API + 依赖注入：

def get_order_repo(db: DbDep) -> SqlAlchemyOrderRepository:
    return SqlAlchemyOrderRepository(db)


def get_order_service(
    repo: Annotated[OrderRepository, Depends(get_order_repo)],
) -> OrderService:
    return OrderService(repo)


OrderServiceDep = Annotated[OrderService, Depends(get_order_service)]


@app.get("/orders/{order_id}", response_model=OrderOut)
def get_order(order_id: int, tenant_id: TenantDep, service: OrderServiceDep) -> OrderOut:
    order = service.get_order(order_id, tenant_id)
    return OrderOut.model_validate(order)

路由函数缩短到几行；404 映射可在路由 except AppError 或全局 handler 完成。

创建订单用例稍复杂，看 Service 如何编排多个 Repository：

class OrderService:
    def __init__(self, orders: OrderRepository, inventory: InventoryRepository) -> None:
        self.orders = orders
        self.inventory = inventory

    def place_order(self, tenant_id: str, lines: list[OrderLineIn]) -> Order:
        for line in lines:
            if not self.inventory.has_stock(line.sku, line.qty):
                raise InsufficientStockError(line.sku, line.qty, self.inventory.available(line.sku))
        order = Order(id=None, tenant_id=tenant_id, total=sum(l.qty * l.price for l in lines))
        order = self.orders.save(order)
        for line in lines:
            self.inventory.deduct(line.sku, line.qty)
        return order

API 层：

@app.post("/orders", response_model=OrderOut, status_code=201)
def place_order(body: PlaceOrderIn, tenant_id: TenantDep, service: OrderServiceDep) -> OrderOut:
    order = service.place_order(tenant_id, body.lines)
    return OrderOut.model_validate(order)

目录怎么摆 ​

一种常见布局（按团队调整）：

myapp/
  api/
    routes/
      orders.py
    schemas/
      order.py
  services/
    order_service.py
  repositories/
    order_repository.py
  models/          # ORM
  domain/          # 纯 dataclass / 领域异常
  deps.py          # get_db, get_*_service
  main.py

api/schemas 是 HTTP 专用 DTO；domain 是 Service 与 Repository 之间传递的结构，避免 ORM 对象 leak 到路由外。小项目可以合并 domain 与 schemas，但 ORM Model 不应直接当 response_model 长期用——字段、循环引用、懒加载会拖累序列化。

模型分化：ORM、BO、DTO 各管一截 ​

课件里「模型分化」要解决的是：一张用户表里既有 password_hash、is_deleted，又有给前端的 email——若到处传同一个类，密码哈希很容易漏进响应。可以拆成三种视角（名字因团队而异）：

示例直觉：ORM User 有 password_hash；BO 可拆 UserIdentity（校验密码）与 UserInfo（邮箱手机）；DTO UserOut 只有 id/username/email。Service 读 ORM → 组 BO → 出 DTO；API 只看见 DTO。不必一上来上完整 DDD，但**「出站禁止直接 dump ORM」**应尽早立规矩。

依赖倒置：Service 不绑死具体仓库 ​

依赖倒置（DIP）：高层（Service）依赖抽象，低层（SQLAlchemy 实现）依赖同一抽象。Python 里常用 Protocol 或 ABC：

from typing import Protocol


class OrderRepository(Protocol):
    def get(self, order_id: int) -> Order | None: ...
    def save(self, order: Order) -> Order: ...


class OrderService:
    def __init__(self, orders: OrderRepository) -> None:
        self._orders = orders

路由通过 Depends 注入 SqlAlchemyOrderRepository；单测注入内存实现。方向是「业务指向接口，细节实现接口」，而不是 Service import 某个具体 session.query 模块。三层文件夹只是表象，**依赖箭头向内（向领域）**才是目标。

事务放哪一层 ​

事务边界通常在 Service 的用例方法上：一个 create_order 里多次 repo.save、扣库存，应同一 session、同一 commit/rollback。Repository 单方法一般只 flush，不随意 commit()，否则 Service 无法组合原子操作。API 层不开事务。

实现方式之一：Service 方法内 with session.begin():；或由 get_db yield 的 Session 在请求末 commit（简单 CRUD 可行，复杂用例仍建议 Service 显式控制）。系列第 16 篇会专讲长短事务抉择。

跨层映射：Schema、Domain、ORM ​

OrderOut.model_validate(order) 从 domain 来；Repository 里 row.to_domain() / OrderModel.from_domain() 集中转换，避免 ORM 懒加载在 API 序列化时爆 N+1。

何时不必强行三层 ​

• 只读健康检查、静态配置：路由直接返回即可。
• CRUD 生成器 / 管理后台原型：可先薄 Service，复杂后再抽。
• 极简单脚本式 API：一层 + 函数也能接受，但要意识到技术债临界点。

一旦同一资源出现「创建 + 状态机 + 通知 + 审计」多条规则，就该收拢到 Service。

胖路由 vs 三层：对照感受 ​

胖路由里常见坏味道：if 嵌套超过三层、同一 session 上 scattered commit、Pydantic 模型与 ORM 模型字段混用、HTTPException 与 return {"error": ...} 并存。 refactor 第一步往往只是把 SQL 挪到 Repository，第二步把 if 规则挪到 Service，路由立刻瘦一圈。不必一天拆完美；按 PR 逐步抽，测试 override 与 Repository 单测会倒逼边界清晰。

多 Service 协作 ​

一个用例可能调 OrderService 与 InventoryService。可以 Orchestrator Service 组合二者，或在 OrderService 构造函数注入 InventoryRepository——避免 Service 互相 import 形成环。跨聚合的最终一致（发消息、邮件）可异步，事务内只做强一致部分。

与依赖注入的配合 ​

三层不是文件夹名字，而是依赖方向：deps.py 里 get_order_service 拼好 Repository 与 Service，路由只拿 OrderServiceDep。换 SQLite 集成测时 override get_db；测 Service 时直接 OrderService(InMemoryOrderRepo())，完全不经 HTTP。分层 + DI 是可测试性的双支柱。

列表、分页与读模型 ​

列表 API 常需 OrderSummaryOut，与详情 OrderOut 不同。Repository 提供 list_summaries(tenant_id, offset, limit) 投影查询，避免 select Order 拉全表字段。分页 count 可单独 count_by_tenant 方法，Service 组装 {items, total}。这些读路径仍遵守分层：路由只调 Service，Service 调 Repository。

防腐层与外部 API ​

调用第三方支付、物流时，adapter 层可视为 Infrastructure，不必强行叫 Repository。Service 依赖 PaymentGateway 接口，与依赖 DB 的 Repository 并列。命名不必教条，关键是向外系统与向数据库的 IO 都不要堆在路由里。

增量 refactor 检查表 ​

• 路由里是否还有 session.query？
• Service 是否 import 了 Request？
• Repository 是否抛 HTTPException？
• 是否同一业务规则出现在两个路由？

任一项为真是下一个 PR 的 refactor 候选。

批量操作与报表 ​

导出 CSV、批量 approve 等读写混合用例，Service 方法显式命名 bulk_ship_orders，内部循环调 Repository 或一条 SQL bulk update，API 只暴露 DTO。报表 SQL 复杂时 Repository 可返回 dataclass /tuple 投影，不必强行 ORM 实体——仍不突破「路由不写 SQL」。

与系列其他篇章的地图 ​

业务逻辑篇：规则进 Service。依赖注入篇：Service 进路由。异常篇：Service 抛 AppError。事务篇：边界在 Service。读完全系列，三层是串联这些横切点的骨架。

常见误解 ​

Repository = 一个表一个类。 可以，也可以按聚合根（Order + OrderLine）一个 Repository；关键是隐藏查询细节，不是机械 1:1。

Service 里写 HTTPException。 业务层抛领域异常，API 或 handler 映射 HTTP（见异常处理篇）。

路由之间互相 import 调内部函数。 应通过 Service 复用逻辑，避免「HTTP 层横向耦合」。

DTO 与 ORM 字段 1:1 forever。 读模型可以 OrderSummaryOut 只含列表需要的列，Repository 提供投影查询。

从单体路由演进到三层 ​

演进路径可以是：第一步提取 Repository，路由仍厚；第二步提取 Service，路由变薄；第三步统一 schemas 与 domain 分离。不必等待「完美领域模型」再拆——先让 SQL 离开路由，收益立竿见影。读已有 Fat API 项目时，用「找 session.query」当 refactor 入口，比抽象讨论架构更落地。

三层拆清楚后，文件上传这类「边界上有二进制、边界外有存储」的能力，就知道该落在 API 哪一段、Service 要不要管、Repository 是否只存 URL——下一篇专门讲 UploadFile 与存储边界。

新成员加入时，用一张「请求穿过哪几层」的 sequence 图讲解，比讲设计模式有效。Code review 问：这个 PR 的新 SQL 是否在 Repository？新 if 业务规则是否在 Service？新 query/body 字段是否在 schemas？三层 checklist 比抽象讨论「是否符合 DDD」更易执行。

分层也便于多人并行：一人改 API schema，一人改 Service 规则，一人改 Repository 索引，冲突少于同文件胖路由。OpenAPI 生成只扫 API 层，不会误暴露 ORM 内部字段——前提是 response_model 用 Out DTO，而不是把 SQLAlchemy model 直接返回。

小结 ​

API 翻译 HTTP，Service 承载规则与事务，Repository 封装持久化。依赖单向向下，DTO 与 ORM 分离。按 PR 渐进抽取，不必一次完美。配合 Depends 注入 Service，配合异常 handler 映射 AppError，分层才真正可测可演进。

反过来看三层：若 Repository 测试很难写，可能是 SQL 散落在 Service；若 Service 测试要 mock 十个对象，可能是用例太大该拆分；若 TestClient 测试要准备二十条 fixture 数据，可能是 API 暴露过宽或缺 factory。分层是诊断工具，不只是文件夹约定。与课件不同的例子（订单、租户、库存）贯穿本系列多篇，读者可按同一 domain 从 DI 读到上传，体会边界如何一步步清晰。

Repository 方法命名用领域语言 find_open_orders_by_tenant，避免 get_data 类模糊名。Service 方法对应用例 cancel_order，与 REST 动词不必一一对应，一个 POST 可能调多个 Service 步骤。

三层不是银弹，但是 FastAPI 后端从 demo 走向可维护服务的最小清晰骨架；与 DI、异常、测试系列文章一起读，效果最佳。

OpenAPI tags 可按 API router 分模块（orders、users），与目录结构一致，文档与代码同构。Service/Repository 不出现在 OpenAPI 中，正是分层对外隐藏实现的方式。

Service 无 FastAPI import，是分层是否干净的快速自检。

当 microservice 拆分时，三层边界往往成为 service 边界：原 monolith 的 OrderService 可演进为独立进程，Repository 换 RPC 客户端，API 层换 BFF。早期分层的投资在拆分时不至于「全局搜索 SQL」式重构。

分层清晰后，onboarding 新人可按 API→Service→Repository 顺序读代码，学习曲线更平。

Repository 返回 domain 对象，不是 dict 或 Row，类型更清晰。

API 层薄、Service 厚、Repository 专，是可持续演进的后端默认形态。

读代码时从 router 往下追 Service、再追 Repository，是理解陌生 FastAPI 项目最快的路径。

配合本系列依赖注入与异常处理两篇阅读，三层结构更容易一次到位。值得在新项目里刻意练习一次。