第十章项目结构与工程化

本章目标：掌握 APIRouter 路由拆分、规范项目目录结构、使用 BaseSettings 管理配置、理解 lifespan 生命周期机制。
预计时长：40 分钟

10.1 APIRouter —— 路由拆分

当项目只有几个接口时，全写在 main.py 里没问题。一旦接口超过 10 个，你需要按功能模块拆分路由。

基本用法

python

# app/routers/users.py
from fastapi import APIRouter

router = APIRouter()


@router.get("/users/")
async def list_users():
    return [{"username": "zhangsan"}, {"username": "lisi"}]


@router.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "username": "zhangsan"}

python

# app/main.py
from fastapi import FastAPI
from app.routers import users

app = FastAPI()

# 注册路由模块
app.include_router(users.router)

prefix、tags、dependencies 参数

APIRouter 和 include_router() 都支持这三个关键参数：

python

# app/routers/users.py
from fastapi import APIRouter, Depends
from app.dependencies.auth import get_current_user

router = APIRouter(
    prefix="/users",          # 所有路由自动加上 /users 前缀
    tags=["用户管理"],         # Swagger UI 中的分组标签
    dependencies=[Depends(get_current_user)],  # 该模块所有接口都需要认证
)


@router.get("/")             # 实际路径: GET /users/
async def list_users():
    return [{"username": "zhangsan"}]


@router.get("/{user_id}")    # 实际路径: GET /users/{user_id}
async def get_user(user_id: int):
    return {"user_id": user_id}

参数	作用	示例
`prefix`	路由前缀，避免每个路由重复写	`prefix="/users"`
`tags`	Swagger UI 分组标签	`tags=["用户管理"]`
`dependencies`	该路由组的公共依赖（如认证）	`dependencies=[Depends(get_current_user)]`
`responses`	统一的错误响应描述	`responses={401: {"description": "未认证"}}`

多模块注册

python

# app/main.py
from fastapi import FastAPI
from app.routers import users, items, auth

app = FastAPI(title="我的项目")

app.include_router(auth.router)
app.include_router(users.router)
app.include_router(items.router)

在 include_router() 时也可以覆盖参数：

python

app.include_router(
    users.router,
    prefix="/api/v1/users",   # 覆盖 router 中的 prefix
    tags=["Users V1"],
)

Swagger UI 效果

路由拆分 + tags 后，Swagger UI 自动按标签分组：

┌─────────────────────────┐
│  认证                    │
│  ├── POST /token         │
│  └── POST /register      │
│                          │
│  用户管理                │
│  ├── GET  /users/        │
│  ├── GET  /users/{id}    │
│  └── PUT  /users/{id}    │
│                          │
│  商品管理                │
│  ├── GET  /items/        │
│  └── POST /items/        │
└─────────────────────────┘

10.2 推荐项目结构

标准目录布局

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口：创建 app、注册路由
│   ├── config.py            # 配置管理：BaseSettings
│   ├── database.py          # 数据库连接与会话管理
│   │
│   ├── models/              # SQLAlchemy ORM 模型
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   │
│   ├── schemas/             # Pydantic 数据模型（请求/响应）
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── item.py
│   │
│   ├── routers/             # 路由模块（APIRouter）
│   │   ├── __init__.py
│   │   ├── auth.py
│   │   ├── users.py
│   │   └── items.py
│   │
│   ├── services/            # 业务逻辑层
│   │   ├── __init__.py
│   │   ├── user_service.py
│   │   └── item_service.py
│   │
│   └── dependencies/        # 公共依赖（认证、数据库会话等）
│       ├── __init__.py
│       ├── auth.py
│       └── database.py
│
├── alembic/                 # 数据库迁移脚本
│   ├── versions/
│   └── env.py
├── alembic.ini
│
├── tests/                   # 测试文件
│   ├── __init__.py
│   ├── test_auth.py
│   ├── test_users.py
│   └── conftest.py
│
├── .env                     # 环境变量（不提交到 Git）
├── .gitignore
├── requirements.txt
└── README.md

各层职责

层	目录	职责	示例
路由层	`routers/`	接收请求、调用服务、返回响应	参数校验、调用 service
服务层	`services/`	业务逻辑	创建用户、计算折扣
模型层	`models/`	数据库表结构	ORM 模型定义
Schema 层	`schemas/`	请求/响应数据结构	Pydantic 模型
依赖层	`dependencies/`	可复用的依赖注入	数据库会话、当前用户

具体文件示例

python

# app/routers/users.py —— 路由层：只做"接"和"转"
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from app.dependencies.database import get_db
from app.dependencies.auth import get_current_user
from app.schemas.user import UserCreate, UserResponse
from app.services import user_service

router = APIRouter(prefix="/users", tags=["用户管理"])


@router.post("/", response_model=UserResponse)
def create_user(user_data: UserCreate, db: Session = Depends(get_db)):
    return user_service.create_user(db, user_data)


@router.get("/me", response_model=UserResponse)
def read_current_user(current_user=Depends(get_current_user)):
    return current_user

python

# app/services/user_service.py —— 服务层：放业务逻辑
from sqlalchemy.orm import Session
from passlib.context import CryptContext

from app.models.user import User
from app.schemas.user import UserCreate

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")


def create_user(db: Session, user_data: UserCreate) -> User:
    hashed_pw = pwd_context.hash(user_data.password)
    db_user = User(username=user_data.username, hashed_password=hashed_pw)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_user_by_username(db: Session, username: str) -> User | None:
    return db.query(User).filter(User.username == username).first()

10.3 配置管理

使用 Pydantic BaseSettings

硬编码配置（如数据库地址、密钥）到代码中是大忌。使用 BaseSettings 统一从环境变量读取：

bash

pip install pydantic-settings

python

# app/config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # 应用配置
    app_name: str = "FastAPI Demo"
    debug: bool = False

    # 数据库
    database_url: str = "sqlite:///./app.db"

    # JWT 认证
    secret_key: str = "change-me-in-production"
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30

    # CORS
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = {
        "env_file": ".env",        # 自动读取 .env 文件
        "env_file_encoding": "utf-8",
    }


settings = Settings()

.env 文件

bash

# .env（项目根目录，不要提交到 Git！）
APP_NAME=我的FastAPI项目
DEBUG=true
DATABASE_URL=postgresql://user:pass@localhost:5432/mydb
SECRET_KEY=a-very-long-random-secret-key-here
ALLOWED_ORIGINS=["http://localhost:3000","http://localhost:5173"]

环境变量映射规则

Python 字段名	对应环境变量名	说明
`database_url`	`DATABASE_URL`	自动转大写 + 下划线
`secret_key`	`SECRET_KEY`	自动匹配
`debug`	`DEBUG`	字符串自动转 `bool`
`access_token_expire_minutes`	`ACCESS_TOKEN_EXPIRE_MINUTES`	字符串自动转 `int`

在代码中使用

python

# app/main.py
from app.config import settings

app = FastAPI(title=settings.app_name, debug=settings.debug)

python

# app/dependencies/auth.py
from app.config import settings

def create_access_token(data: dict):
    return jwt.encode(data, settings.secret_key, algorithm=settings.algorithm)

优势：

开发环境用 .env，生产环境用真实环境变量，代码不用改
类型自动转换（字符串 → int / bool / list）
配置字段有类型检查和默认值

10.4 启动事件与生命周期

lifespan 上下文管理器（推荐方式）

FastAPI 旧版的 @app.on_event("startup") 和 @app.on_event("shutdown") 已废弃。现在推荐使用 lifespan：

python

from contextlib import asynccontextmanager
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    # ======== 启动时执行 ========
    print("应用启动：初始化资源...")
    # 例如：创建数据库连接池、加载 ML 模型、初始化缓存
    db_pool = await create_db_pool()
    app.state.db_pool = db_pool

    yield  # <-- 应用运行中

    # ======== 关闭时执行 ========
    print("应用关闭：清理资源...")
    await db_pool.close()


app = FastAPI(lifespan=lifespan)

执行流程

应用启动
  ↓
lifespan() 进入 → 执行 yield 前的代码（初始化）
  ↓
yield → 应用正常运行，处理请求
  ↓
收到关闭信号（Ctrl+C）
  ↓
lifespan() 继续 → 执行 yield 后的代码（清理）
  ↓
应用退出

实际应用示例

python

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import httpx


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时创建 HTTP 客户端（复用连接池，性能更好）
    app.state.http_client = httpx.AsyncClient(timeout=30)
    print("HTTP 客户端已创建")

    yield

    # 关闭时释放资源
    await app.state.http_client.aclose()
    print("HTTP 客户端已关闭")


app = FastAPI(lifespan=lifespan)


@app.get("/proxy")
async def proxy_request(request: Request):
    """使用共享的 HTTP 客户端调用外部 API"""
    client: httpx.AsyncClient = request.app.state.http_client
    resp = await client.get("https://httpbin.org/get")
    return resp.json()

lifespan vs on_event 对比

特性	`lifespan`（推荐）	`on_event`（已废弃）
资源共享	✅ 通过 `app.state`	✅ 通过全局变量
启动 + 关闭在同一函数	✅	❌ 分散在两个函数
上下文管理器模式	✅ 确保资源释放	❌ 关闭回调可能不执行
FastAPI 官方推荐	✅	❌

10.5 动手练习

练习 1：路由拆分

将以下单文件代码拆分为多模块结构：

python

# 原始 main.py（所有代码都在一个文件）
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/")
def list_users():
    return [{"name": "zhangsan"}]

@app.get("/items/")
def list_items():
    return [{"name": "手机"}]

@app.get("/orders/")
def list_orders():
    return [{"id": 1, "item": "手机"}]

拆分为：

app/
├── main.py
└── routers/
    ├── __init__.py
    ├── users.py
    ├── items.py
    └── orders.py

练习 2：配置管理

创建 app/config.py，定义以下配置项并从 .env 读取：

app_name（str，默认 "My App"）
database_url（str，默认 "sqlite:///./test.db"）
debug（bool，默认 False）
api_prefix（str，默认 "/api/v1"）

创建对应的 .env 文件，在 main.py 中使用配置。

练习 3：lifespan 实践

创建一个 lifespan，在启动时打印配置信息，在关闭时打印运行统计：

python

from contextlib import asynccontextmanager
from datetime import datetime

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.start_time = datetime.now()
    print(f"应用启动于 {app.state.start_time}")
    print(f"配置：{settings.app_name}, Debug={settings.debug}")

    yield

    uptime = datetime.now() - app.state.start_time
    print(f"应用已运行 {uptime.total_seconds():.0f} 秒")

本章小结

概念	要点
`APIRouter`	按功能模块拆分路由，支持 `prefix`、`tags`、`dependencies`
`include_router()`	在 `main.py` 中注册路由模块
项目结构	routers → services → models/schemas，各层职责清晰
`BaseSettings`	从环境变量 / `.env` 读取配置，自动类型转换
`lifespan`	替代已废弃的 `on_event`，用上下文管理器管理启动/关闭逻辑
`app.state`	在 lifespan 中存储共享资源，路由中通过 `request.app.state` 访问

下一章预告：代码写完了，怎么保证它是对的？下一章我们将学习如何用 TestClient 编写自动化测试，用依赖覆盖 Mock 数据库，以及异步测试方案。

第十章 项目结构与工程化 ​

10.1 APIRouter —— 路由拆分 ​

基本用法 ​

prefix、tags、dependencies 参数 ​

多模块注册 ​

Swagger UI 效果 ​

10.2 推荐项目结构 ​

标准目录布局 ​

各层职责 ​

具体文件示例 ​

10.3 配置管理 ​

使用 Pydantic BaseSettings ​

.env 文件 ​

环境变量映射规则 ​

在代码中使用 ​

10.4 启动事件与生命周期 ​

lifespan 上下文管理器（推荐方式） ​

执行流程 ​

实际应用示例 ​

lifespan vs on_event 对比 ​

10.5 动手练习 ​

练习 1：路由拆分 ​

练习 2：配置管理 ​

练习 3：lifespan 实践 ​

本章小结 ​

第十章项目结构与工程化

10.1 APIRouter —— 路由拆分

基本用法

prefix、tags、dependencies 参数

多模块注册

Swagger UI 效果

10.2 推荐项目结构

标准目录布局

各层职责

具体文件示例

10.3 配置管理

使用 Pydantic BaseSettings

.env 文件

环境变量映射规则

在代码中使用

10.4 启动事件与生命周期

lifespan 上下文管理器（推荐方式）

执行流程

实际应用示例

lifespan vs on_event 对比

10.5 动手练习

练习 1：路由拆分

练习 2：配置管理

练习 3：lifespan 实践

本章小结