本节目标：把「能跑的单文件 FastAPI」重构成可维护的分层应用，讲清 router / service / repository 的职责边界、
APIRouter的拆分装配、Depends的请求级与进程级依赖，以及lru_cache单例的用法与陷阱。
适用版本：Python 3.12+（实测 3.14.6）；FastAPI 0.143.0、Starlette 1.7.0、httpx 0.28.1
5.1 FastAPI 应用结构与依赖注入
第 4 章我们把测试、性能回归和覆盖率门禁立了起来。从本章起进入后端工程：先把应用骨架搭对，后面的数据库、缓存、认证才有地方挂。这一节不讲「FastAPI 怎么用」——那在入门篇已经打过底；我们讲真实项目里怎么组织代码，以及依赖注入到底在解决什么问题。
5.1.1 单文件 app 的尽头
几乎每个 FastAPI 项目都从这样一份 main.py 起步：所有路由、业务逻辑、数据存取挤在一个文件里。它能跑到几十个接口，然后开始失控：
- 改一个业务规则要在几百行里翻找，冲突不断；
- 想给「创建订单」写单元测试，必须先起 HTTP 层；
- 数据库连接、配置对象在模块顶层
new出来，测试之间互相污染。
根因是职责混在一起。解法是老生常谈但确实有效的分层：把「收请求」「定规则」「读写数据」三件事拆到不同层，每层只依赖它下面那层。本节用一个商品服务把它落到可运行的代码上。
5.1.2 四层：api / service / repository / schema
先定边界，再写代码。四层的职责与依赖方向如下：
| 层 | 目录 | 职责 | 允许依赖 |
|---|---|---|---|
| api（路由层） | app/api/ | 解析请求、调用 service、把异常翻译成 HTTP 状态码 | service、schema |
| service（业务层） | app/services/ | 业务规则、编排、事务边界 | repository |
| repository（数据层） | app/repositories/ | 纯数据存取，不含业务判断 | 无（或 ORM） |
| schema（契约层） | app/schemas.py | 出入参的 Pydantic 模型 | 无 |
依赖方向必须单向：api → service → repository。反过来的依赖（service 里 raise HTTPException、repository 里判断权限）是分层腐化的开始。
数据层用内存字典即可，重点是接口形态；第 6 章会把它换成 SQLAlchemy 2.1.4，service 层几乎不用改：
# app/repositories/items.py
from dataclasses import dataclass
@dataclass
class Item:
id: int
name: str
price: float
class ItemRepository:
"""最底层：只负责数据存取，不含业务规则。"""
def __init__(self) -> None:
self._rows: dict[int, Item] = {}
self._next_id = 1
def add(self, name: str, price: float) -> Item:
item = Item(id=self._next_id, name=name, price=price)
self._rows[item.id] = item
self._next_id += 1
return item
def get(self, item_id: int) -> Item | None:
return self._rows.get(item_id)
def list(self) -> list[Item]:
return list(self._rows.values())
service 层定义业务异常，不碰 HTTP——这是关键纪律：
# app/services/items.py
from app.repositories.items import Item, ItemRepository
class ItemNotFound(Exception):
"""业务异常，由 API 层翻译成 HTTP 状态码。"""
class ItemService:
def __init__(self, repo: ItemRepository) -> None:
self._repo = repo
def create(self, name: str, price: float) -> Item:
return self._repo.add(name=name, price=price)
def get(self, item_id: int) -> Item:
item = self._repo.get(item_id)
if item is None:
raise ItemNotFound(item_id)
return item
def list(self) -> list[Item]:
return self._repo.list()
ItemService 只认 ItemRepository，不认识 FastAPI。于是「测试业务规则」可以完全脱离 HTTP：
repo = ItemRepository()
svc = ItemService(repo)
svc.create("x", 1.0)
print("service sees repo row:", repo.list()[0].name)
service sees repo row: x
5.1.3 APIRouter 拆分与装配
路由层按资源拆成多个文件，每个文件导出一个 APIRouter。注意 HTTPException 只在这一层出现——把 service 抛出的 ItemNotFound 翻译成 404：
# app/api/items.py
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from app.deps import get_service
from app.schemas import ItemIn, ItemOut
from app.services.items import ItemNotFound, ItemService
router = APIRouter(prefix="/items", tags=["items"])
@router.post("", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
def create_item(
payload: ItemIn,
svc: Annotated[ItemService, Depends(get_service)],
) -> ItemOut:
item = svc.create(name=payload.name, price=payload.price)
return ItemOut(id=item.id, name=item.name, price=item.price)
@router.get("/{item_id}", response_model=ItemOut)
def get_item(
item_id: int,
svc: Annotated[ItemService, Depends(get_service)],
) -> ItemOut:
try:
item = svc.get(item_id)
except ItemNotFound:
raise HTTPException(status_code=404, detail=f"item {item_id} not found") from None
return ItemOut(id=item.id, name=item.name, price=item.price)
main.py 只做装配——把 router 挂到应用上，版本前缀在这一层统一：
# app/main.py
from fastapi import FastAPI
from app.api.items import router as items_router
app = FastAPI(title="Shop API", version="0.1.0")
app.include_router(items_router, prefix="/api/v1")
prefix 分层很实用：路由文件里写相对路径 /items，版本号 /api/v1 只在装配处出现一次。将来出 v2，新写一个 router 用不同前缀即可，业务代码零改动。
5.1.4 Depends 的层级：请求级 vs 进程级
Depends 常被当成「参数默认值」用，其实它的价值在于控制对象的生命周期。两种粒度要分清：
- 进程级单例：配置、连接池、仓库实例——整个应用一份，启动时建好，请求间共享。
- 请求级对象：service、当前用户、数据库会话——每个请求一份，用完即弃。
把它们放进 app/deps.py，靠组合拼出层级：
# app/deps.py
from functools import lru_cache
from fastapi import Depends
from app.repositories.items import ItemRepository
from app.services.items import ItemService
@lru_cache(maxsize=1)
def get_repository() -> ItemRepository:
"""进程级单例：整个应用共享一个仓库实例。"""
return ItemRepository()
def get_service(repo: ItemRepository = Depends(get_repository)) -> ItemService:
"""请求级依赖：每次请求构造一个 service，复用单例 repo。"""
return ItemService(repo)
依赖可以嵌套：get_service 依赖 get_repository，FastAPI 会按拓扑序依次解析，且同一请求内对同一依赖只解析一次（默认缓存）。这样路由里只声明它真正需要的 get_service，repo 自动注入进来。
5.1.5 lru_cache 单例：好用但会咬人
@lru_cache(maxsize=1) 是最省事的进程级单例写法——第一次调用建对象，之后返回同一个。实测两次解析拿到同一实例：
from app.deps import get_repository
print("repo singleton:", get_repository() is get_repository())
repo singleton: True
但它有两个坑，团队里几乎都踩过：
- 参数必须可哈希。
lru_cache按参数做键；如果依赖函数带不可哈希参数（如list、未冻结的dict），会直接TypeError。 - 测试之间不隔离。单例一旦建好就常驻进程，上一个用例写进去的数据会漏给下一个。测试里要么调用
get_repository.cache_clear()，要么用下面的覆盖机制。
生产里配置对象也常用这个模式（配合 3.1 节的 pydantic-settings）。若需要更明确的「只初始化一次」语义，可用模块级模块变量 + functools.cache，或直接放进 lifespan（5.3 节）。
5.1.6 用 dependency_overrides 做测试隔离
FastAPI 提供了官方的替换开关：app.dependency_overrides[原依赖] = 替身。测试时把单例仓库换成隔离实例，用完清空：
from fastapi.testclient import TestClient
from app.deps import get_repository
from app.main import app
from app.repositories.items import ItemRepository
def make_client() -> TestClient:
fake = ItemRepository()
fake.add("测试商品", 10.0)
app.dependency_overrides[get_repository] = lambda: fake
return TestClient(app)
client = make_client()
print("overridden repo:", client.get("/api/v1/items").json())
app.dependency_overrides.clear()
print("real repo after clear:", client.get("/api/v1/items").json())
overridden repo: [{'id': 1, 'name': '测试商品', 'price': 10.0}]
real repo after clear: []
替换的是依赖解析本身，业务代码一行不动——这正是把依赖写进 Depends 而不是在函数体里 import 单例的回报。
5.1.7 跑一遍：真实请求与响应
用一个脚本把正常路径、404、校验失败都打一遍。注意 FastAPI 0.143.0 / Starlette 1.7.0 下，TestClient 会打印一条弃用提示「httpx with starlette.testclient is deprecated」——功能不受影响；异步测试建议直接用 httpx.ASGITransport（本节末尾给出）。
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
print("POST:", *client.post("/api/v1/items", json={"name": "键盘", "price": 199.0}).json().values())
print("GET /items:", client.get("/api/v1/items").status_code)
print("GET /items/999:", client.get("/api/v1/items/999").status_code)
print("invalid:", client.post("/api/v1/items", json={"name": "", "price": -1}).status_code)
真实输出（版本号来自本机实测）：
POST /items: 1 键盘 199.0
GET /items: 200
GET /items/999: 404
POST invalid: 422
201 由 status_code=HTTP_201_CREATED 声明；404 是 service 异常被翻译而来；422 是 Pydantic 校验自动拦截——三层各司其职，互不越界。
不想被 TestClient 的弃用提示打扰，可以改用异步客户端（httpx.ASGITransport 直接驱动 ASGI 应用，无需真实端口）：
import asyncio
import httpx
from app.main import app
async def main() -> None:
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
r = await c.post("/api/v1/items", json={"name": "键盘", "price": 199.0})
print("POST:", r.status_code, r.json())
asyncio.run(main())
POST: 201 {'id': 1, 'name': '键盘', 'price': 199.0}
注意 ASGITransport 不会触发 lifespan——需要启动/关闭钩子时用 asgi_lifespan 或回到 TestClient 的上下文管理器，这正是 5.3 节的主题。
5.1.8 目录树与延伸
app/
├── main.py # 只做装配
├── deps.py # 依赖定义（生命周期）
├── schemas.py # Pydantic 契约
├── api/items.py # 路由层
├── services/items.py # 业务层
└── repositories/items.py # 数据层
这套分层的收益会在后面几章持续兑现：6 章把 ItemRepository 换成 SQLAlchemy 会话、8 章在 service 里挂缓存、9 章在 deps.py 加认证依赖——变化都被关在单层内。
小结
- 分层不是仪式：api / service / repository / schema 各管一段，依赖方向单向，
HTTPException只在路由层出现。 APIRouter按资源拆分，main.py只做装配，版本前缀在装配处统一。Depends的本质是控制生命周期：进程级单例（配置、连接池）与请求级对象（service、会话）分开定义、组合拼装。@lru_cache(maxsize=1)是最省事的单例，但要注意参数可哈希、以及测试间不隔离这两个坑。dependency_overrides是官方测试隔离手段，替换依赖解析而不动业务代码。TestClient在本机 Starlette 1.7.0 下会打印 httpx 弃用提示（功能正常）；异步场景改用httpx.ASGITransport，但它不跑lifespan。
到这里应用有了骨架，但「一个请求从进门到出门」之间还缺少横切的环节——日志、请求 ID、耗时统计、鉴权预检。下一节我们把这些统一收进中间件与请求上下文。
延伸阅读：Python Web 框架全景 、Python 微服务架构 。
阅读导航：上一节：属性测试、性能回归与覆盖率门禁 · 下一节：路由、中间件与请求上下文 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。