FastAPI 大项目该怎么组织？APIRouter 和分层怎么划分？

简化版

FastAPI 官方脚手架只给了单文件示例，项目一大就需要自己定结构，业界收敛出的做法是**「按业务域拆包 + 每个域内部分层」：顶层是 app/api/（路由）、app/schemas/（Pydantic 模型）、app/models/（ORM 模型）、app/services/ 或 crud/（业务逻辑）、app/core/（配置、安全、日志）、app/deps.py（可复用依赖）——小项目按「技术角色」分目录就够，项目变大后应该改成「按业务域分包」（app/users/{router,schemas,models,service}.py），因为改一个功能时相关文件都在同一个目录里**。路由靠 APIRouter 拆分：每个模块一个 router，用 prefix/tags/dependencies/responses 声明公共部分，再在 main.py 里 include_router() 汇总；版本化就是再套一层 router（/api/v1）。四条关键约定：① Pydantic 模型和 ORM 模型必须分开（UserIn/UserOut/UserInDB vs User 表模型），混用会导致「数据库字段一改 API 就变、密码字段泄露到响应里」；② 业务逻辑放 service 层，路由只做「解析参数 → 调 service → 返回」，这样同一逻辑能被路由、CLI、定时任务、消息消费者复用；③ 依赖用 Annotated 定义可复用别名（DbDep = Annotated[AsyncSession, Depends(get_db)]），大项目能省掉成百上千行重复；④ 配置用 pydantic-settings 集中管理并用 lru_cache 缓存。还有一个高频问题是循环导入——解法是「models 不 import schemas、service 不 import router」的单向依赖，以及必要时用 TYPE_CHECKING。核心记忆：小项目按角色分、大项目按业务域分；APIRouter 的 prefix/tags/dependencies；schema 和 model 必须分开；逻辑放 service，路由只做转发。

详细版

两种目录组织方式对比：

	按技术角色分（小项目）	按业务域分（大项目）
结构	`routers/`、`schemas/`、`models/`	`users/`、`orders/`、`products/`
找文件	要在 4 个目录间跳	同一个功能都在一个目录
适合规模	< 20 个路由	模块多、多人协作
边界	模糊	清晰（可拆微服务）
缺点	大了之后目录很长	小项目显得啰嗦

★ ① 小项目（按技术角色）★
app/
├── main.py               # ★创建 app、include_router、中间件★
├── core/
│   ├── config.py         # ★Settings（pydantic-settings）★
│   ├── security.py       # 密码哈希、JWT
│   └── logging.py
├── api/
│   ├── deps.py           # ★★可复用依赖（get_db / get_current_user）★★
│   └── v1/
│       ├── __init__.py   # ★汇总 router★
│       ├── users.py
│       └── items.py
├── models/               # ★SQLAlchemy ORM★
├── schemas/              # ★★Pydantic（和 models 分开）★★
├── services/             # ★业务逻辑★
├── db/
│   ├── base.py           # Base、metadata
│   └── session.py        # engine、SessionLocal
└── tests/

★ ② 大项目（按业务域）★
app/
├── main.py
├── core/                 # ★跨域共享★
│   ├── config.py  security.py  logging.py  exceptions.py
├── db/
├── users/                # ★★一个业务域一个包★★
│   ├── router.py
│   ├── schemas.py
│   ├── models.py
│   ├── service.py
│   ├── deps.py           # ★该域专属依赖★
│   └── exceptions.py
├── orders/
│   └── （同样结构）
└── shared/               # ★跨域复用的东西★

# ③ ★APIRouter 的完整用法★
from fastapi import APIRouter, Depends, HTTPException, status

router = APIRouter(
    prefix="/users",                          # ★统一前缀★
    tags=["users"],                           # ★文档分组★
    dependencies=[Depends(verify_api_key)],   # ★★路由级依赖（不需要返回值）★★
    responses={404: {"description": "未找到"}},# ★文档里的公共响应★
)

@router.get("/{user_id}", response_model=UserOut)
async def get_user(user_id: int, svc: UserServiceDep):
    return await svc.get(user_id)

# ④ ★汇总：api/v1/__init__.py★
from fastapi import APIRouter
from .users import router as users_router
from .items import router as items_router

api_router = APIRouter()
api_router.include_router(users_router)
api_router.include_router(items_router)

# ⑤ ★main.py★
from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db(); await init_redis()        # ★启动★
    yield
    await close_redis()                        # ★关闭★

def create_app() -> FastAPI:                   # ★★工厂函数（便于测试）★★
    app = FastAPI(title=settings.APP_NAME, lifespan=lifespan,
                  docs_url="/docs" if settings.DEBUG else None)  # ★生产可关文档★
    app.add_middleware(CORSMiddleware, allow_origins=settings.CORS_ORIGINS)
    app.include_router(api_router, prefix="/api/v1")   # ★★版本前缀在这里加★★
    register_exception_handlers(app)
    return app

app = create_app()

# ⑥ ★★Annotated 依赖别名（大项目必备）★★
# api/deps.py
from typing import Annotated
from fastapi import Depends

DbDep = Annotated[AsyncSession, Depends(get_async_session)]
CurrentUser = Annotated[User, Depends(get_current_user)]
AdminUser = Annotated[User, Depends(get_current_admin)]
SettingsDep = Annotated[Settings, Depends(get_settings)]

# 使用：签名瞬间变干净
async def create_order(data: OrderIn, db: DbDep, user: CurrentUser):
    ...

# ⑦ ★schema 分离（★必须★）★
class UserBase(BaseModel):
    email: str
    name: str
class UserCreate(UserBase):                    # ★入参★
    password: str                              # ★★只进不出★★
class UserUpdate(BaseModel):                   # ★PATCH 用，全可选★
    name: str | None = None
class UserOut(UserBase):                       # ★★出参：没有 password★★
    model_config = ConfigDict(from_attributes=True)
    id: int
    created_at: datetime
class UserInDB(UserOut):                       # ★内部用★
    hashed_password: str

⚠️ 三个必须记住的点：① Pydantic 模型（schema）和 ORM 模型（model）必须分开，而且输入输出还要再分。直接把 ORM 模型当响应模型返回，会带来三个问题：数据库字段一改 API 契约就跟着变（本该是两件独立的事）、敏感字段容易泄露（hashed_password、is_deleted、内部备注字段会一并序列化出去）、无法表达「创建时必填但更新时可选」这类差异。标准做法是一组 schema：UserBase（公共字段）、UserCreate（含 password，只进不出）、UserUpdate（全部可选，给 PATCH）、UserOut（响应，不含密码）、UserInDB（内部流转，含 hashed_password）。② 路由函数只做「解析参数 → 调 service → 返回」，业务逻辑放 service 层。把逻辑写在路由里的代价是：无法被 CLI 命令、Celery 任务、消息消费者、另一个接口复用；测试必须走 HTTP（慢且难构造边界场景）；换框架或加一个 GraphQL 入口时要重写。判断标准很简单——路由函数超过 20 行就该往下沉。③ 用 Annotated 定义依赖别名。DbDep = Annotated[AsyncSession, Depends(get_db)] 之后，几百个路由的签名从 db: AsyncSession = Depends(get_db) 变成 db: DbDep——不只是短，更重要的是改依赖实现时只改一处，而且函数默认值位置是干净的，路由函数可以脱离 FastAPI 被直接调用和单测。

完整版教学

一、两种目录组织方式

★ ★方式一：按技术角色（技术分层）★
  app/routers/users.py       app/routers/orders.py
  app/schemas/user.py        app/schemas/order.py
  app/models/user.py         app/models/order.py
  app/services/user.py       app/services/order.py
  ★ ✓ 优点：★分层清晰、新人一眼看懂★、小项目很自然
  ★ ✗ 缺点：★改一个功能要在 4 个目录间跳★
            ★目录里文件越来越多（20 个业务 = 每个目录 20 个文件）★
            ★业务边界模糊（哪些文件属于"订单"？）★

★ ★方式二：按业务域（垂直切分）★
  app/users/{router,schemas,models,service,deps}.py
  app/orders/{router,schemas,models,service,deps}.py
  ★ ✓ 优点：★★改一个功能所有文件都在一个目录★★
            ★边界清晰 → 以后拆微服务只要搬一个目录★
            ★多人协作时冲突少（各改各的目录）★
  ★ ✗ 缺点：★小项目显得啰嗦★、跨域复用要额外规划

★ ★怎么选（★实际判断★）★：
  ┌────────────────────────────┬────────────────────┐
  │ 路由 < 20 个、1~2 人        │ ★按技术角色★        │
  │ 业务模块多、多人协作         │ ★★按业务域★★        │
  │ 将来可能拆微服务             │ ★按业务域★          │
  │ 已有代码是角色分且没痛点     │ ★别为了重构而重构★  │
  └────────────────────────────┴────────────────────┘
  ★ ★经验：从技术角色开始，感到"跳来跳去很烦"时再切换★

★ ★无论哪种，都要有的几个东西★：
  ① ★core/config.py★     —— 集中配置
  ② ★deps.py★            —— 可复用依赖
  ③ ★core/exceptions.py★ —— 自定义异常 + 处理器
  ④ ★db/session.py★      —— 引擎和会话
  ⑤ ★main.py 的 create_app() 工厂★

★ ★★依赖方向必须单向（防循环导入）★★：
  ┌──────────────────────────────────────────────┐
  │   router  →  service  →  model / db           │
  │      ↓          ↓                             │
  │   schema    schema                            │
  │                                               │
  │ ★禁止：model → schema、service → router★       │
  └──────────────────────────────────────────────┘
  ★ 循环导入的常见来源：
    ① ★models 之间互相引用外键关系★
       ✓ 用字符串引用：relationship("Order")
    ② ★schema 引用 model 的枚举★
       ✓ 枚举单独放 core/enums.py
    ③ ★service 之间互相调用★
       ✓ 抽出更底层的共享 service，或用事件解耦
    ④ ★TYPE_CHECKING 只为类型注解导入★
       from typing import TYPE_CHECKING
       if TYPE_CHECKING:
           from .models import User        # ★运行时不导入★

两种组织方式各有适用规模：按技术角色分层清晰、小项目很自然，但改一个功能要在 4 个目录间跳、业务边界模糊；按业务域则是改一个功能所有文件都在一个目录、边界清晰（以后拆微服务只要搬一个目录）、多人协作冲突少。经验是从技术角色开始，感到「跳来跳去很烦」时再切换——别为了重构而重构。无论哪种都要有 core/config.py、deps.py、core/exceptions.py、db/session.py 和 create_app() 工厂。依赖方向必须单向（router → service → model，禁止 model → schema 和 service → router），循环导入的四个常见来源是：models 互相引用外键（用字符串 relationship("Order")）、schema 引用 model 的枚举（枚举单独放 core/enums.py）、service 互相调用、以及只为类型注解的导入（用 TYPE_CHECKING）。

二、APIRouter 的组织

★ ★APIRouter 的五个参数★：
  router = APIRouter(
      prefix="/users",                       # ★★不能以 / 结尾★★
      tags=["users"],                        # ★Swagger 里的分组★
      dependencies=[Depends(verify_token)],  # ★★对该 router 下所有路由生效★★
      responses={404: {"model": ErrorOut}},  # ★文档用★
      deprecated=False,
  )

★ ★router 级 dependencies 的特点★：
  dependencies=[Depends(verify_api_key)]
  ★ ✓ ★不需要返回值★（纯粹做校验/副作用）
  ★ ✓ ★所有路由自动生效★，不用每个函数写参数
  ★ ✓ 典型用途：★鉴权、限流、审计日志、租户校验★
  ★ ✗ 但★路由函数拿不到它的返回值★
    → 需要值就还得在函数签名里再写一次 Depends

★ ★三层嵌套（★版本化的标准做法★）★：
  # users.py
  router = APIRouter(prefix="/users", tags=["users"])
  # api/v1/__init__.py
  api_v1 = APIRouter()
  api_v1.include_router(users.router)
  api_v1.include_router(orders.router)
  # main.py
  app.include_router(api_v1, prefix="/api/v1")
  → ★最终路径：/api/v1/users/{id}★
  ★ ★好处：版本前缀只在一处，改版本不用改各模块★

★ ★include_router 时还能追加★：
  app.include_router(
      users.router,
      prefix="/admin",                   # ★★叠加在 router 自己的 prefix 前★★
      tags=["admin"],                    # ★★追加而不是替换 tags★★
      dependencies=[Depends(require_admin)],  # ★★追加依赖★★
  )
  ★ ★同一个 router 可以挂载多次★（不同前缀、不同依赖）
    app.include_router(users.router, prefix="/api/v1")
    app.include_router(users.router, prefix="/api/latest")

★ ★路由顺序的坑★：
  @router.get("/users/me")        # ★必须写在前面★
  @router.get("/users/{user_id}")
  ★ FastAPI ★按注册顺序匹配★（★和 Flask/Werkzeug 不同！★）
  → ★如果 /users/{user_id} 在前，访问 /users/me 会把 "me" 当 user_id★
  → ★然后 int 转换失败报 422★
  ★ ★记忆：具体路径写在参数路径前面★

★ ★tags 与文档组织★：
  app = FastAPI(openapi_tags=[
      {"name": "users", "description": "用户管理"},
      {"name": "orders", "description": "订单相关",
       "externalDocs": {"url": "https://..."}},
  ])
  ★ 控制 Swagger 里的分组顺序和说明

★ ★把路由从文档里隐藏★：
  @router.get("/internal", include_in_schema=False)   # ★内部接口★
  ★ 或整个 router：APIRouter(include_in_schema=False)

★ ★大项目的自动注册（可选）★：
  import pkgutil, importlib
  def auto_include(app, package):
      for _, name, _ in pkgutil.iter_modules(package.__path__):
          mod = importlib.import_module(f"{package.__name__}.{name}")
          if hasattr(mod, "router"):
              app.include_router(mod.router)
  ★ ✓ 新增模块不用改 main.py
  ★ ✗ ★隐式魔法，排查时不好找★ → ★小心使用★

APIRouter 的五个参数里最有价值的是 dependencies——它对该 router 下所有路由自动生效且不需要返回值，典型用途是鉴权、限流、审计、租户校验；但路由函数拿不到它的返回值，需要值就还得在签名里再写一次。版本化的标准做法是三层嵌套：模块 router → 版本 router → app，版本前缀只在一处，改版本不用动各模块。include_router 时还能追加 prefix、tags 和 dependencies，所以同一个 router 能挂载多次。有个和 Flask 完全不同的坑：FastAPI 按注册顺序匹配路由——/users/me 必须写在 /users/{user_id} 前面，否则 "me" 会被当成 user_id 然后 int 转换失败报 422。记忆：具体路径写在参数路径前面。

三、分层与 service

★ ★三层职责★：
  ┌──────────────────────────────────────────────────────┐
  │ ★路由层（router）★                                     │
  │   - 解析和校验参数（Pydantic 自动做）                    │
  │   - ★鉴权（依赖）★                                      │
  │   - 调用 service                                       │
  │   - ★把业务异常转成 HTTP 状态码★                        │
  │   ★不写业务逻辑、不直接查数据库★                         │
  ├──────────────────────────────────────────────────────┤
  │ ★服务层（service）★                                    │
  │   - ★业务规则、事务边界、编排多个仓储★                   │
  │   - ★抛业务异常（不是 HTTPException）★                  │
  │   ★不认识 HTTP、不认识 Request★                         │
  ├──────────────────────────────────────────────────────┤
  │ ★数据层（repository / crud / model）★                  │
  │   - ★纯粹的数据存取★                                   │
  │   - 不含业务判断                                        │
  └──────────────────────────────────────────────────────┘

★ ★反例：逻辑写在路由里★
  @router.post("/orders")
  async def create_order(data: OrderIn, db: DbDep, user: CurrentUser):
      # ★★80 行业务逻辑★★
      product = await db.get(Product, data.product_id)
      if not product: raise HTTPException(404)
      if product.stock < data.qty: raise HTTPException(400, "库存不足")
      ...计算价格、优惠、创建订单、扣库存、发消息...
  ★ 代价：
    ✗ ★CLI/定时任务/消息消费者无法复用★
    ✗ ★测试必须走 HTTP★（构造边界场景很麻烦）
    ✗ ★加一个 GraphQL 入口要重写★
    ✗ ★事务边界不清晰★

★ ✓ ★正确：路由只做转发★
  # service/order.py
  class OrderService:
      def __init__(self, db: AsyncSession):
          self.db = db
      async def create(self, data: OrderCreate, user: User) -> Order:
          product = await self.db.get(Product, data.product_id)
          if product is None:
              raise NotFoundError("商品不存在")            # ★★业务异常★★
          if product.stock < data.qty:
              raise InsufficientStock(available=product.stock)
          ...
          return order

  # router
  @router.post("/orders", response_model=OrderOut, status_code=201)
  async def create_order(data: OrderIn, svc: OrderServiceDep,
                         user: CurrentUser):
      return await svc.create(data, user)          # ★★三行搞定★★

★ ★业务异常 → HTTP 的转换（★关键设计★）★：
  # core/exceptions.py
  class AppError(Exception):
      code, status, message = "app_error", 400, "错误"
  class NotFoundError(AppError):
      code, status = "not_found", 404
  class InsufficientStock(AppError):
      code, status = "insufficient_stock", 409

  # 注册处理器
  @app.exception_handler(AppError)
  async def app_error_handler(request, exc: AppError):
      return JSONResponse(status_code=exc.status,
                          content={"error": {"code": exc.code,
                                             "message": exc.message}})
  ★ ★好处：service 层完全不认识 HTTP★
    → ★同一个 service 能被 CLI/Celery 调用，异常照样有意义★

★ ★service 怎么注入★：
  # ① 函数依赖
  def get_order_service(db: DbDep) -> OrderService:
      return OrderService(db)
  OrderServiceDep = Annotated[OrderService, Depends(get_order_service)]

  # ② 类本身作为依赖（★FastAPI 支持★）
  class OrderService:
      def __init__(self, db: DbDep):     # ★★构造参数也能用 Depends★★
          self.db = db
  OrderServiceDep = Annotated[OrderService, Depends()]   # ★Depends() 空参★

★ ★要不要 repository 层★：
  ✓ 需要：★换数据库/需要 mock 数据层/领域复杂★
  ✗ 不需要：★小项目里 SQLAlchemy 本身就是仓储★
  ★ ★别为了"架构完整"加一层什么都不做的 CRUD★
    （典型反模式：crud.get(db, id) 只是 db.get(Model, id) 的包装）

三层职责要划清：路由层只做「解析参数 → 鉴权 → 调 service → 把业务异常转成 HTTP」，不写业务逻辑、不直接查数据库；服务层管业务规则和事务边界，抛业务异常而不是 HTTPException，且完全不认识 HTTP 和 Request；数据层只做纯粹的存取。「service 抛业务异常」是关键设计——定义 AppError 体系并注册 exception_handler 统一转成 HTTP 响应，好处是同一个 service 能被 CLI 和 Celery 调用，异常照样有意义。service 的注入有两种写法，其中 FastAPI 支持把类本身作为依赖（构造函数参数也能用 Depends）。最后一个判断：别为了「架构完整」加一层什么都不做的 CRUD 包装——crud.get(db, id) 只是 db.get(Model, id) 的转发就是典型的反模式，只有在需要换数据库、mock 数据层或领域确实复杂时才加 repository。

四、配置与依赖管理

★ ★配置：pydantic-settings★：
  # core/config.py
  from pydantic_settings import BaseSettings, SettingsConfigDict
  from functools import lru_cache

  class Settings(BaseSettings):
      model_config = SettingsConfigDict(
          env_file=".env", env_file_encoding="utf-8",
          extra="ignore",                    # ★忽略无关的环境变量★
          case_sensitive=False,
      )
      APP_NAME: str = "MyAPI"
      DEBUG: bool = False
      SECRET_KEY: str                        # ★★没有默认值 = 必须提供★★
      DATABASE_URL: str
      REDIS_URL: str = "redis://localhost:6379/0"
      CORS_ORIGINS: list[str] = []           # ★环境变量里写 JSON 数组★
      ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

      @computed_field
      @property
      def async_database_url(self) -> str:
          return self.DATABASE_URL.replace("postgresql://",
                                           "postgresql+asyncpg://")

  ★@lru_cache★                                # ★★只解析一次★★
  def get_settings() -> Settings:
      return Settings()

  settings = get_settings()

★ ★为什么用 lru_cache★：
  ① ★Settings() 每次会读 .env 和环境变量（有 IO）★
  ② ★作为依赖时可以被 dependency_overrides 覆盖★：
     SettingsDep = Annotated[Settings, Depends(get_settings)]
     # 测试里
     app.dependency_overrides[get_settings] = lambda: TestSettings()
  ★ ★比直接用模块级全局 settings 更好测★

★ ★多环境配置★：
  # ① 不同的 .env 文件
  ENV=production python -m app     # ★读 .env.production★
  model_config = SettingsConfigDict(
      env_file=(".env", f".env.{os.getenv('ENV', 'dev')}"))
  # ② 或子类
  class DevSettings(Settings): DEBUG = True
  class ProdSettings(Settings): DEBUG = False
  ★ ★生产环境用真实环境变量（容器 env / k8s secret），别依赖 .env 文件★

★ ★★依赖的组织（Annotated 别名）★★：
  # api/deps.py
  from typing import Annotated
  from fastapi import Depends

  # ★基础★
  DbDep = Annotated[AsyncSession, Depends(get_async_session)]
  SettingsDep = Annotated[Settings, Depends(get_settings)]
  RedisDep = Annotated[Redis, Depends(get_redis)]

  # ★认证（★分层次★）★
  CurrentUser = Annotated[User, Depends(get_current_user)]
  ActiveUser = Annotated[User, Depends(get_current_active_user)]
  AdminUser = Annotated[User, Depends(get_current_admin)]

  # ★分页★
  class PageParams(BaseModel):
      page: int = Field(1, ge=1)
      size: int = Field(20, ge=1, le=100)     # ★★上限必须有★★
  PageDep = Annotated[PageParams, Depends()]

  # ★service★
  OrderServiceDep = Annotated[OrderService, Depends(get_order_service)]

  # ★使用：签名极其干净★
  @router.get("/orders")
  async def list_orders(page: PageDep, user: CurrentUser,
                        svc: OrderServiceDep):
      return await svc.list(user, page)

★ ★依赖的层次复用★：
  def get_current_user(token: str = Depends(oauth2_scheme),
                       db: DbDep) -> User: ...
  def get_current_active_user(user: CurrentUser) -> User:
      if not user.is_active: raise HTTPException(403, "账号已禁用")
      return user
  def get_current_admin(user: ActiveUser) -> User:
      if not user.is_admin: raise HTTPException(403, "需要管理员权限")
      return user
  ★ ★依赖可以依赖依赖 → 形成校验链★
  ★ FastAPI 会★缓存同一请求内的相同依赖★（use_cache=True 默认）
    → ★get_current_user 在一个请求里只执行一次★

配置用 pydantic-settings 并配 lru_cache——后者不只是为了性能（Settings() 每次要读 .env 和环境变量），更重要的是把它做成依赖后可以被 dependency_overrides 覆盖，比模块级全局变量好测。没有默认值的字段等于「必须提供」（缺失时启动直接失败，这是好事）；生产环境要用真实的环境变量而不是 .env 文件。依赖组织的关键是用 Annotated 定义别名：基础依赖（DbDep/SettingsDep/RedisDep）、分层次的认证依赖（CurrentUser → ActiveUser → AdminUser，依赖可以依赖依赖，形成校验链）、分页参数、service——之后路由签名会极其干净。还要知道 FastAPI 会缓存同一请求内的相同依赖（use_cache=True 默认），所以 get_current_user 在一个请求里只执行一次。

五、循环导入与工程细节

★ ★循环导入的四种典型场景与解法★：

  ① ★ORM 模型互相引用★
     # models/user.py 需要 Order，models/order.py 需要 User
     ✓ ★关系用字符串★：
       class User(Base):
           orders = relationship("Order", back_populates="user")
       class Order(Base):
           user = relationship("User", back_populates="orders")
     ✓ ★类型注解用 TYPE_CHECKING★：
       if TYPE_CHECKING:
           from .order import Order
       orders: Mapped[list["Order"]] = relationship(...)

  ② ★schema 之间嵌套引用★
     class UserOut(BaseModel):
         orders: list["OrderOut"] = []
     ★ Pydantic V2 会自动解析前向引用；必要时 model_rebuild()

  ③ ★service 互相调用★
     OrderService 需要 UserService，UserService 也需要 OrderService
     ✓ ★抽出更底层的共享服务★
     ✓ ★或用领域事件解耦★（订单创建后发事件，用户服务订阅）
     ✓ ★或在方法内部 import★（下策但有效）

  ④ ★deps 和 service 互相引用★
     ✓ ★deps 只负责"构造"，不含逻辑★
     ✓ 把共享类型放 core/types.py

★ ★main.py 用工厂函数的理由★：
  def create_app() -> FastAPI: ...
  app = create_app()
  ★ ✓ ★测试里能创建多个独立的 app★（不同配置）
  ★ ✓ ★避免 import 时的副作用★
  ★ ✓ 便于按环境装配不同的中间件

★ ★启动与关闭：lifespan★：
  @asynccontextmanager
  async def lifespan(app: FastAPI):
      # ★启动★
      app.state.redis = await create_redis_pool(settings.REDIS_URL)
      app.state.http = httpx.AsyncClient(timeout=10)
      yield
      # ★关闭（★优雅退出很重要★）★
      await app.state.http.aclose()
      await app.state.redis.close()
  ★ ✗ 老的 @app.on_event("startup") ★已弃用★
  ★ ★app.state 上的东西要通过依赖暴露，别在路由里直接用★：
    async def get_redis(request: Request) -> Redis:
        return request.app.state.redis

★ ★数据库会话的依赖（异步）★：
  async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
      async with AsyncSessionLocal() as session:
          try:
              yield session
              # ★★注意：不要在这里 commit★★
          except Exception:
              await session.rollback()
              raise
  ★ ★事务边界应该在 service 层★（一个业务操作 = 一个事务）
  ★ ✗ 反例：在依赖里自动 commit → ★service 无法控制事务★

★ ★静态检查与规范★：
  □ ★ruff（lint + format，替代 flake8/black/isort）★
  □ ★mypy 或 pyright★（★FastAPI 项目类型注解本来就全★）
  □ ★pre-commit 钩子★
  □ ★pytest + coverage★
  # pyproject.toml
  [tool.ruff.lint]
  select = ["E", "F", "I", "N", "UP", "B", "S"]   # ★S = 安全检查★

★ ★目录之外的工程文件★：
  pyproject.toml        # ★依赖 + 工具配置（替代 setup.py/requirements）★
  .env.example          # ★★提交这个，不提交 .env★★
  Dockerfile / docker-compose.yml
  alembic/              # 数据库迁移
  Makefile 或 justfile  # ★常用命令（run/test/lint/migrate）★

循环导入有四种典型场景：ORM 模型互相引用（关系用字符串 relationship("Order")，类型注解用 TYPE_CHECKING）、schema 嵌套引用（Pydantic V2 会自动解析前向引用）、service 互相调用（抽出共享服务或用领域事件解耦）、deps 与 service 互引。main.py 用工厂函数的理由是「测试里能创建多个独立的 app、避免 import 副作用」。lifespan 取代了已弃用的 @app.on_event，而且放在 app.state 上的资源要通过依赖暴露（get_redis(request)），别在路由里直接访问。数据库会话依赖里不要 commit——事务边界应该在 service 层（一个业务操作对应一个事务），在依赖里自动 commit 会让 service 失去事务控制权。工具链推荐 ruff（一个工具替代 flake8/black/isort）+ mypy + pre-commit。

六、实践清单

★ ★推荐的起步结构（中等项目）★：
  app/
  ├── main.py                # create_app()
  ├── core/
  │   ├── config.py          # ★Settings + get_settings（lru_cache）★
  │   ├── security.py        # 密码、JWT
  │   ├── exceptions.py      # ★AppError 体系 + handlers★
  │   └── logging.py
  ├── db/
  │   ├── base.py            # Base（所有 model 都 import 它）
  │   └── session.py         # engine + AsyncSessionLocal + get_session
  ├── api/
  │   ├── deps.py            # ★★所有 Annotated 别名★★
  │   └── v1/
  │       ├── __init__.py    # api_router 汇总
  │       ├── users.py
  │       └── orders.py
  ├── models/                # SQLAlchemy
  ├── schemas/               # ★Pydantic（In/Out/Update 分开）★
  ├── services/              # 业务逻辑
  └── tests/
      ├── conftest.py
      └── api/

★ 检查清单：
  □ ★schema 和 ORM model 分开，In/Out 也分开★
  □ ★UserOut 里没有 password/hashed_password★
  □ ★路由函数 < 20 行，逻辑在 service★
  □ ★service 抛业务异常，不抛 HTTPException★
  □ ★用 Annotated 定义依赖别名★
  □ ★版本前缀只在 main.py 加一次★
  □ ★/users/me 写在 /users/{id} 前面★
  □ ★分页 size 有上限★
  □ ★Settings 用 lru_cache 且可被 override★
  □ ★lifespan 里创建的资源有对应的关闭★
  □ ★事务边界在 service 不在依赖★
  □ ★依赖方向单向（无循环导入）★
  □ ★.env 不提交，提交 .env.example★
  □ ★生产关闭 /docs（或加鉴权）★

★ ★何时该拆微服务（★别过早★）★：
  ✗ ★"代码变多了"不是理由★
  ✓ ★团队大到互相阻塞部署★
  ✓ ★不同模块的伸缩需求差异巨大★
  ✓ ★技术栈必须不同★
  ★ ★按业务域组织的单体（模块化单体）已经能解决 80% 的问题★
    → ★而且随时可以拆★

★ 一句话总结：
  ★"小项目按技术角色分目录，模块多了改成按业务域分包；
    APIRouter 用 prefix/tags/dependencies 声明公共部分，
    版本前缀只在 main.py 加一次；
    schema 和 ORM model 必须分开、In/Out 也要分开；
    业务逻辑放 service（抛业务异常不抛 HTTPException），
    路由只做转发；依赖用 Annotated 定义别名。"★

检查清单里最关键的五条：UserOut 里没有密码字段、路由函数少于 20 行、service 抛业务异常而不是 HTTPException、/users/me 写在 /users/{id} 前面、事务边界在 service 不在依赖。最后一条判断很重要：「代码变多了」不是拆微服务的理由——按业务域组织的模块化单体已经能解决 80% 的问题，而且随时可以拆；真正的信号是「团队大到互相阻塞部署」「不同模块的伸缩需求差异巨大」。

记忆钩子：「FastAPI 大项目的组织有两种：★小项目按技术角色分目录（routers/schemas/models/services），项目变大后改成按业务域分包（users/orders 各自一个包）★——后者的价值是★改一个功能所有文件都在一个目录、边界清晰到以后拆微服务只要搬一个目录★；★经验是从技术角色开始，感到『跳来跳去很烦』时再切换★。★APIRouter 的核心是 prefix/tags/dependencies★，其中 ★router 级 dependencies 对下面所有路由生效且不需要返回值★（适合鉴权限流审计），★但路由函数拿不到它的返回值★。★版本化用三层嵌套：模块 router → 版本 router → app.include_router(prefix=‘/api/v1’)★，版本前缀只在一处。★有个和 Flask 完全不同的坑：FastAPI 按注册顺序匹配路由★——★/users/me 必须写在 /users/{user_id} 前面★，否则 ‘me’ 会被当 user_id 然后 422。★四条关键约定★：★① Pydantic schema 和 ORM model 必须分开，且 In/Out 再分★（UserBase/UserCreate 含 password 只进不出/UserUpdate 全可选给 PATCH/UserOut 不含密码/UserInDB 含 hashed_password）——混用会导致★数据库字段一改 API 契约就变、敏感字段泄露到响应★；★② 路由只做『解析参数→鉴权→调 service→返回』，超过 20 行就该下沉★，★service 抛业务异常（AppError 体系）而不是 HTTPException★，这样★同一逻辑能被 CLI/Celery/消息消费者复用★，再用 exception_handler 统一转成 HTTP；★③ 依赖用 Annotated 定义别名★（DbDep/CurrentUser/AdminUser/PageDep/OrderServiceDep），★依赖可以依赖依赖形成校验链（CurrentUser→ActiveUser→AdminUser）★，且 ★FastAPI 会缓存同一请求内的相同依赖所以只执行一次★；★④ 配置用 pydantic-settings + lru_cache★（不只为性能，更是★为了能被 dependency_overrides 覆盖★），★没有默认值的字段=必须提供★。★循环导入四解法★：ORM 关系用★字符串 relationship(‘Order’)★、类型注解用 ★TYPE_CHECKING★、service 互调抽共享服务或用领域事件、依赖方向严格单向（★禁止 model→schema、service→router★）。其他工程点：★main.py 用 create_app() 工厂★（测试能造多个 app）、★lifespan 取代已弃用的 @app.on_event★、★app.state 上的资源要通过依赖暴露★、★数据库会话依赖里不要 commit（事务边界在 service）★、★分页 size 必须有上限★、★生产关闭 /docs★。最后：★『代码变多』不是拆微服务的理由，模块化单体已能解决 80% 的问题且随时可拆★。」

七、常见误区与追问

误区：ORM 模型可以直接当 response_model 用，省得再写一套 Pydantic 模型。 三个问题会接踵而至。① 敏感字段泄露——User 表里有 hashed_password、is_deleted、internal_note、last_login_ip，直接序列化会把它们全部返回给客户端；等你发现时可能已经泄露很久了（而且 hashed_password 泄露意味着攻击者可以离线爆破）。② API 契约和数据库结构被绑死——数据库加个字段、改个字段名，API 的响应结构就跟着变了，而这两件事本该是独立演进的；客户端也会因为「后端加了个内部字段」而收到意外的数据。③ 无法表达输入输出的差异——创建用户时需要 password（明文，只进不出）、响应时需要 id 和 created_at（只出不进）、PATCH 时所有字段都该可选——一套模型表达不了这三种形态。标准做法是一组 schema：UserBase（公共字段）、UserCreate、UserUpdate、UserOut、UserInDB，通过继承减少重复。
误区：业务逻辑写在路由函数里最直观，反正也能跑。 短期确实最快，但会在四个地方付出代价。① 无法复用——同样的「创建订单」逻辑，后台管理要用、CLI 批量导入要用、Celery 补偿任务要用、消息消费者也要用，写在路由里就只能复制或者用 TestClient 自己调自己（荒谬但真实发生过）。② 测试成本高——测一个「库存不足时应该报错」的分支，走 HTTP 要准备认证、构造请求体、断言状态码；而如果是 service 层的函数，直接 pytest.raises(InsufficientStock) 就完了。③ 事务边界不清晰——路由里散落着多个 await db.commit()，很难看出哪些操作应该是原子的。④ 换入口要重写——加一个 GraphQL 端点或 gRPC 服务时，业务逻辑得整个搬一遍。判断标准很实用：路由函数超过 20 行、或者出现了 if 嵌套的业务判断，就该往 service 层沉。
误区：service 层里直接 raise HTTPException(404) 很方便，反正最终也是返回 HTTP。 这让 service 层被 HTTP 协议污染了。后果是：① service 无法在非 HTTP 场景复用——Celery 任务里抛 HTTPException 毫无意义，日志里看到「404 Not Found」也很困惑；② 状态码和业务语义耦合——「库存不足」这个业务事实，在不同场景可能对应 409、422 或者只是一条警告，写死在 service 里就没法灵活处理；③ 单元测试要 import FastAPI。正确做法是定义自己的异常体系：class AppError(Exception) 带上 code（业务错误码）、status（建议的 HTTP 状态）、message，各业务异常继承它；然后在应用层注册一个 @app.exception_handler(AppError) 统一转成 JSON 响应。这样 service 层完全不认识 HTTP，而 HTTP 层只需要一个处理器——CLI 和 Celery 里捕获同样的异常也能得到有意义的信息。
误区：路由的注册顺序不重要，FastAPI 会自动选最匹配的。 FastAPI（Starlette）是按注册顺序从上到下匹配的，第一个匹配上的胜出——这和 Flask/Werkzeug「按规则复杂度排序」的行为完全相反。所以如果你先注册了 @router.get("/users/{user_id}") 再注册 @router.get("/users/me")，那么访问 /users/me 时会先匹配上前者，把 "me" 当作 user_id——如果参数类型是 int，会得到一个令人困惑的 422「Input should be a valid integer」；如果是 str，则会拿着 "me" 去查数据库然后 404。规则是：具体路径（静态段）一定要写在参数路径前面。同类的还有 /files/latest vs /files/{name}、/orders/export vs /orders/{id}。排查这类问题时，打印 [r.path for r in app.routes] 看实际顺序最直接。
误区：项目一大就该拆微服务。 「代码变多」本身从来不是拆分的理由，而拆分的代价却是实实在在的：网络调用替代了函数调用（延迟、超时、重试、序列化）、分布式事务（本地事务变成了 saga 或最终一致）、部署和监控复杂度翻倍、本地开发要起一堆服务、跨服务的重构变得极其困难。真正的拆分信号只有三个：① 团队大到互相阻塞（多个团队改同一个仓库、发布要排队协调）；② 伸缩需求差异巨大（一个模块要 50 个副本、另一个 2 个就够，还挤在一个进程里很浪费）；③ 技术栈必须不同（某个模块要用 Go 或需要 GPU）。在此之前，「按业务域组织的模块化单体」能解决 80% 的问题——各业务域有清晰的目录边界、只通过明确的接口互相调用、数据表按域划分，而且当真的需要拆时，只要把一个目录搬出去改成 HTTP 调用即可。这条路径远比一上来就拆微服务稳妥。
追问：Annotated 依赖别名在大项目里到底能带来什么？ 四点实际收益。① 消除重复——一个几百个路由的项目里，db: AsyncSession = Depends(get_async_session) 这行会重复几百次；定义 DbDep = Annotated[AsyncSession, Depends(get_async_session)] 之后变成 db: DbDep，而且换实现时只改一处（比如从同步 Session 迁移到 AsyncSession，或者给 get_db 加上租户隔离逻辑）。② 函数签名保持干净——旧写法把 Depends(...) 塞在默认值位置，导致这个函数无法脱离 FastAPI 被直接调用（直接调用时 db 会是一个 Depends 对象）；Annotated 写法的默认值位置是空的或者是真正的默认值，路由函数可以当普通函数单测。③ 表达依赖的层次——CurrentUser / ActiveUser / AdminUser 三个别名一眼就能看出权限要求，比读 Depends(get_current_admin) 更直观。④ 类型检查友好——mypy/pyright 能正确推断参数类型（旧写法下默认值类型和注解不匹配，需要特殊处理）。实践上建议把所有别名集中放在 api/deps.py，作为「这个项目有哪些可注入的东西」的清单。
追问：get_settings 为什么要用 lru_cache 而不是直接一个模块级的 settings 变量？ 两个理由，第二个更重要。① 性能——Settings() 每次实例化都会读取 .env 文件、扫描环境变量、执行所有字段的校验，虽然单次很快，但如果它被当作依赖在每个请求里调用，累积起来就是浪费；@lru_cache 保证整个进程只解析一次。② 可测试性——把它包装成函数并作为依赖使用（SettingsDep = Annotated[Settings, Depends(get_settings)]）之后，测试里可以用 app.dependency_overrides[get_settings] = lambda: TestSettings() 整体替换配置——比如把数据库指向测试库、把 bcrypt rounds 调到 4、关掉限流和缓存。如果代码里到处直接 from app.core.config import settings 用模块级全局变量，就只能用 monkeypatch 逐个属性打补丁（脆弱且啰嗦）。注意用了 lru_cache 后，测试里改环境变量不会生效——需要 get_settings.cache_clear()。折中做法是「模块级 settings 供 import 时就需要的地方用（比如 create_app 里配置中间件），依赖注入的 get_settings 供路由和 service 用」。
追问：数据库会话的依赖里到底该不该 commit？ 不该，事务边界应该由 service 层控制。常见的错误写法是在依赖里写 yield session 之后跟一个 await session.commit()——这看起来省事（业务代码不用管提交），但破坏了事务语义：① service 无法把多个操作组合成一个原子单元（比如「扣库存 + 创建订单 + 记录流水」必须同生共死，如果每一步都由框架自动提交就做不到）；② 出错时的回滚时机不明确——异常抛出后依赖里的 commit 可能已经执行了一部分；③ 无法表达「只读事务」。正确的分工是：依赖只负责「创建会话、异常时回滚、最后关闭」（async with AsyncSessionLocal() as session: yield session），而 commit() 由 service 层在完成一个完整业务操作后显式调用——service 最清楚哪些步骤应该原子化。更进一步的做法是提供一个 unit_of_work 上下文管理器，让事务边界在代码里显式可见。另外注意 yield 之后的清理代码在响应发送之后执行，所以不要在那里做耗时操作。

八、加强记忆

FastAPI 大项目的组织有两种：小项目按技术角色分目录（routers/、schemas/、models/、services/），项目变大后改成按业务域分包（users/、orders/ 各自一个包）——后者的价值是改一个功能所有文件都在一个目录、边界清晰到以后拆微服务只要搬一个目录；经验是从技术角色开始，感到「跳来跳去很烦」时再切换。APIRouter 的核心是 prefix/tags/dependencies，其中 router 级 dependencies 对下面所有路由生效且不需要返回值（适合鉴权、限流、审计），但路由函数拿不到它的返回值。版本化用三层嵌套：模块 router → 版本 router → app.include_router(prefix="/api/v1")，版本前缀只在一处。有个和 Flask 完全不同的坑：FastAPI 按注册顺序匹配路由——/users/me 必须写在 /users/{user_id} 前面，否则 "me" 会被当成 user_id 然后报 422。四条关键约定：① Pydantic schema 和 ORM model 必须分开，且输入输出再分（UserBase、UserCreate 含 password 只进不出、UserUpdate 全可选给 PATCH、UserOut 不含密码、UserInDB 含 hashed_password）——混用会导致数据库字段一改 API 契约就变、敏感字段泄露到响应里；② 路由只做「解析参数 → 鉴权 → 调 service → 返回」，超过 20 行就该下沉，service 抛业务异常（AppError 体系）而不是 HTTPException，这样同一份逻辑能被 CLI、Celery、消息消费者复用，再用 exception_handler 统一转成 HTTP 响应；③ 依赖用 Annotated 定义别名（DbDep、CurrentUser、AdminUser、PageDep、OrderServiceDep），依赖可以依赖依赖形成校验链（CurrentUser → ActiveUser → AdminUser），而且 FastAPI 会缓存同一请求内的相同依赖，所以只执行一次；④ 配置用 pydantic-settings + lru_cache（不只为性能，更是为了能被 dependency_overrides 覆盖），没有默认值的字段等于「必须提供」。循环导入的四个解法：ORM 关系用字符串 relationship("Order")、类型注解用 TYPE_CHECKING、service 互相调用时抽出共享服务或用领域事件、以及保持依赖方向严格单向（禁止 model → schema、service → router）。其他工程要点：main.py 用 create_app() 工厂（测试能创建多个独立 app）、lifespan 取代已弃用的 @app.on_event、app.state 上的资源要通过依赖暴露、数据库会话依赖里不要 commit（事务边界在 service）、分页 size 必须有上限、生产环境关闭 /docs。最后记住：「代码变多」不是拆微服务的理由，模块化单体已能解决 80% 的问题且随时可拆。

简化版

详细版