Skip to content

第 21 章 · 配置管理与应用生命周期

本章目标:用 pydantic-settings 管理环境变量配置,用 lifespan 优雅地管理应用级资源的启动与释放。

21.1 为什么需要配置管理

数据库地址、密钥、第三方凭据这些值:会随环境变化(开发/测试/生产不同),而且往往敏感、不能写进代码仓库。业界通行做法是放进环境变量,由应用在运行时读取。

但环境变量有个天然缺陷——它永远是字符串

python
import os
items_per_user = os.getenv("ITEMS_PER_USER")   # "50" 而不是 50
# 类型转换和校验都得自己手写,漏一处就是线上事故

21.2 pydantic-settings:带校验的配置

pydantic-settings 把 Pydantic 的类型转换与校验能力带到了环境变量上:

bash
uv add pydantic-settings python-dotenv
python
# config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "Awesome API"       # 有默认值则可缺省
    admin_email: str
    items_per_user: int = 50            # 自动从字符串转成 int 并校验
    database_url: str = "sqlite:///./app.db"

    # 从项目根目录 .env 文件读取(需 python-dotenv)
    model_config = SettingsConfigDict(env_file=".env")


settings = Settings()  # 实例化时读取环境变量,大小写不敏感
bash
# .env(记得加入 .gitignore!)
ADMIN_EMAIL="admin@example.com"
APP_NAME="ChimichangApp"

实例化时按 环境变量 > .env 文件 > 默认值 的优先级取值;环境变量名与字段名大小写不敏感匹配(APP_NAMEapp_name)。启动服务时临时覆盖:

bash
ADMIN_EMAIL="deadpool@example.com" APP_NAME="ChimichangApp" fastapi run main.py

配置放独立模块

Settings 放进 config.py(配合第 19 章的包结构),各处 from .config import settings 即可。注意 .env 绝不能提交到 Git。

21.3 Settings 作为依赖 + @lru_cache 单例

直接用全局 settings 对象的问题:测试时难以替换。更优雅的方式是把配置做成依赖:

python
# config.py —— 注意这次不创建全局实例
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    app_name: str = "Awesome API"
    admin_email: str
    items_per_user: int = 50

    model_config = SettingsConfigDict(env_file=".env")
python
# main.py
from functools import lru_cache
from fastapi import Depends, FastAPI
from .config import Settings

app = FastAPI()


@lru_cache                # 关键:整个进程只创建一次 Settings
def get_settings():
    return Settings()


@app.get("/info")
def info(settings: Settings = Depends(get_settings)):
    return {"app_name": settings.app_name}

为什么需要 @lru_cache?如果只有 def get_settings(): return Settings()每次请求都会重新实例化 Settings、重新读一遍磁盘上的 .env 文件——文件 IO 是慢操作,完全没必要。加上 @lru_cache 后首次调用执行并缓存结果,之后所有请求直接复用同一个对象。

而它作为依赖的最大红利是测试友好

python
# test_main.py
def test_info_with_test_settings():
    def test_settings():
        return Settings(admin_email="test@example.com", _env_file=None)

    app.dependency_overrides[get_settings] = test_settings
    resp = client.get("/info")
    assert resp.json()["admin_email"] == "test@example.com"
    app.dependency_overrides.clear()

21.4 lifespan:应用生命周期钩子

有些资源属于整个应用而非单个请求:数据库连接池、ML 模型、后台任务队列。它们应该"启动时创建一次、关闭时释放"。现代 FastAPI 用 lifespan 参数实现——一个带 yield 的异步上下文管理器:

python
import contextlib
from fastapi import FastAPI, Request
from sqlalchemy.ext.asyncio import create_async_engine


@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    # ===== yield 之前:startup,应用开始收请求之前执行一次 =====
    engine = create_async_engine(
        settings.database_url, pool_size=10, max_overflow=20,
    )
    app.state.engine = engine          # 挂到 app.state 上供各处使用
    ml_models = {"model": load_expensive_model()}  # 模拟加载大模型
    app.state.models = ml_models

    yield                               # ← 应用在此期间处理所有请求

    # ===== yield 之后:shutdown,应用停止接收请求后执行一次 =====
    await engine.dispose()             # 释放连接池
    del ml_models["model"]             # 释放显存/内存


app = FastAPI(lifespan=lifespan)


@app.get("/predict/{x}")
async def predict(x: float, request: Request):
    model = request.app.state.models["model"]  # 每个请求复用同一模型
    return {"result": model.predict([x])}

原理:@asynccontextmanager 把函数变成异步上下文管理器——进入 with 块前执行 yield 前半段,退出时执行后半段。FastAPI 接管了这个 with:启动时进入,关闭时退出。

已废弃的 on_event

旧教程里的 @app.on_event("startup") / @app.on_event("shutdown") 已废弃。且两者互斥:一旦传入 lifespan 参数,startup/shutdown 事件处理器将不再被调用。新项目一律使用 lifespan

另外注意区分:lifespan 管应用级资源(进程一份);第 11 章 Depends(yield)请求级资源(每个请求一份)。

21.5 本章小结

  • 环境变量恒为字符串,pydantic-settings 提供类型转换+校验+默认值+.env 读取的一体化方案;
  • 取值优先级:真实环境变量 > .env > 字段默认值;.env 必须加入 .gitignore
  • 配置做成依赖(get_settings)并用 @lru_cache 保证全进程只实例化一次、只读一次盘;
  • 依赖化的配置在测试中可用 dependency_overrides 秒换测试配置;
  • 应用级资源用 FastAPI(lifespan=...) + @asynccontextmanager 管理:yield 前 startup、yield 后 shutdown;on_event 已废弃。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. pydantic-settings 读取配置值的优先级顺序是?

2. get_settings() 上加 @lru_cache 的核心目的是?

3. 关于 lifespan,下列说法错误的是?

4. 数据库连接池这类"全应用共享一份"的资源,正确的初始化位置是?

🛠️ 动手实践

  1. 为你的 todos 应用建立 config.pydatabase_urldebugmax_page_size 三个配置项,从 .env 读取并在路由中通过 Depends(get_settings) 使用。
  2. 写一个测试:override get_settingsmax_page_size 改为 1,断言分页接口最多返回 1 条数据。
  3. 给应用的 lifespan 加上"启动时预热缓存、关闭时清空"逻辑,用 with TestClient(app) 写测试验证 startup 确实执行了。

一切就绪,最后把应用真正送上生产环境:第 22 章 · 生产部署实战