1. 入口
FastAPI 配置分为两层。pyproject.toml 中的 [tool.fastapi] 配置 CLI 如何找到应用;应用 Settings 则配置创建出来的服务如何运行。
当前项目使用:
1[tool.fastapi]2entrypoint = "app.main:app"
冒号左侧是模块路径,右侧是模块中的对象名:
1app.main:app2│ │ └── app/main.py 中的 FastAPI 对象3│ └────── app/main.py 模块4└────────── app Python 包
对应代码是:
1from app.factory import create_app23app = create_app()
FastAPI CLI 执行的效果相当于:
1from app.main import app
配置入口后,本地开发不必重复传文件路径:
1uv run fastapi dev2uv run fastapi run
fastapi dev 面向开发,会启用自动重载;fastapi run 面向生产启动。[tool.fastapi] 不会决定监听端口、worker 数量或环境变量,这些仍由 CLI 参数和运行环境控制。
需要特别注意,fastapi dev 会在导入应用前尝试设置 FASTAPI_ENV=development,而 fastapi run 不会自动设置生产值。上一篇的 Settings 读取的是 APP__ENVIRONMENT,并不会把 FASTAPI_ENV 自动转换成 settings.app.environment,因此部署时仍应显式设置 APP__ENVIRONMENT=production。
两条命令的默认监听地址也不同:fastapi dev 使用 127.0.0.1,只供本机访问;fastapi run 使用 0.0.0.0,可能被网络中的其他设备访问。生产部署还需要结合反向代理、HTTPS、防火墙和进程管理,不能把执行一条 fastapi run 当作完整的部署方案。
如果项目改成 src 布局,例如应用位于 src/agent_api/main.py,入口也必须同步调整:
1[tool.fastapi]2entrypoint = "agent_api.main:app"
这个写法要求 agent_api 已经按照 src 布局安装到当前 Python 环境中。只是在仓库里创建 src/agent_api 目录,却没有正确配置并安装项目时,Python 仍然可能找不到这个包。
入口配置错误时,FastAPI CLI 会在启动阶段报告模块无法导入或找不到 app,它与请求路由返回 404 是不同问题。
2. 应用工厂
不要在 main.py 中读取几十个环境变量,再逐个创建中间件、数据库和模型。应用工厂可以把组装过程集中起来:
01from collections.abc import Callable02from contextlib import AbstractAsyncContextManager0304from fastapi import FastAPI0506from app.api.router import router as api_router07from app.core.settings import Settings, get_settings08from app.middleware import register_middleware091011def create_app(settings: Settings | None = None) -> FastAPI:12current_settings = settings if settings is not None else get_settings()1314app = FastAPI(15title=current_settings.app.name,16version="0.1.0",17debug=current_settings.app.environment == "development",18)19app.state.settings = current_settings2021register_middleware(app, current_settings)22app.include_router(api_router)23return app
settings 参数允许测试显式传入配置;正常启动则调用 get_settings()。这比在测试中修改全局环境变量后重新导入整个模块更可控。
app.state.settings 把当前配置挂在应用实例上,适合需要从 Request 找到应用资源的依赖。它不是让所有业务代码直接访问全局状态的理由,业务服务仍应通过构造参数接收自己需要的配置。
Settings 可以在进程中缓存:
1from functools import lru_cache234@lru_cache(maxsize=1)5def get_settings() -> Settings:6return Settings()
缓存后,同一 worker 进程只读取和校验一次配置。部署多个 worker 时,每个进程都有自己的 Settings 和资源,Python 内存不会跨进程共享。
修改 .env 或 TOML 后,缓存对象不会自行变化。生产环境应重启或滚动更新应用,而不是尝试在请求中清除缓存。
3. 依赖
FastAPI 依赖可以把 Settings 提供给路由:
01from typing import Annotated0203from fastapi import Depends, Request0405from app.core.settings import Settings060708def get_current_settings(request: Request) -> Settings:09return request.app.state.settings101112SettingsDep = Annotated[Settings, Depends(get_current_settings)]
get_settings() 负责在创建应用时读取配置,get_current_settings() 则负责把这个已经确定的对象交给路由。名称看起来相近,但职责不同。这样测试向应用工厂传入 Settings 后,路由、中间件和 lifespan 都会看到同一个实例。
路由只声明需要配置,不负责读取配置文件:
01from fastapi import APIRouter0203from app.api.dependencies import SettingsDep0405router = APIRouter()060708@router.get("/runtime-info")09async def runtime_info(settings: SettingsDep) -> dict[str, str]:10return {11"app_name": settings.app.name,12"environment": settings.app.environment,13"model": settings.model.name,14}
真实接口不一定应该公开模型名称和环境信息。这里用于演示依赖传递;生产健康检查只返回排障所需的非敏感字段。
依赖注入的价值不是少写一行导入,而是让对象来源明确且可以替换。对于 Settings,应用工厂参数就是最直接的替换入口:
1from app.factory import create_app234def test_factory_uses_injected_settings(test_settings):5app = create_app(test_settings)67assert app.state.settings is test_settings
这里的 test_settings fixture 会在第 6 节定义。模型客户端等资源依赖仍然可以使用 app.dependency_overrides 替换,但覆盖完成后应及时清空,避免影响其他测试。
如果路由自己调用 Settings(),测试就很难替换来源,还可能在每次请求中重复读取文件。
4. 生命周期
Settings 是轻量配置对象,数据库连接池、异步 HTTP 客户端和 LangGraph checkpointer 则是需要创建与关闭的长期资源。它们适合放进 FastAPI lifespan,也就是从应用启动到应用停止的这段生命周期:
01from collections.abc import AsyncIterator02from contextlib import asynccontextmanager0304from fastapi import FastAPI05from httpx import AsyncClient, Timeout0607from app.agent.graph import create_graph08from app.agent.model import create_model09from app.core.checkpointer import open_checkpointer10from app.core.settings import Settings111213def create_lifespan(settings: Settings):14@asynccontextmanager15async def lifespan(app: FastAPI) -> AsyncIterator[None]:16async with AsyncClient(17timeout=Timeout(settings.model.timeout_seconds),18) as http_client:19async with open_checkpointer(settings) as checkpointer:20model = create_model(settings)21graph = create_graph(22model=model,23checkpointer=checkpointer,24)2526app.state.http_client = http_client27app.state.model = model28app.state.checkpointer = checkpointer29app.state.graph = graph3031yield3233return lifespan
create_lifespan() 是一个工厂:外层函数先接收 Settings,内层 lifespan() 再由 FastAPI 调用。yield 之前负责创建资源,yield 之后会按照 async with 的逆序自动关闭资源。
如果某个资源在 yield 之前创建失败,应用不会开始接收请求,已经进入的 async with 也会负责清理。这比先启动服务、等到请求到来后才发现核心资源不可用更容易排查。
示例中的 create_model()、create_graph() 和 open_checkpointer() 是项目自己的工厂函数,后续 LangChain/LangGraph 文章会实现。其中,open_checkpointer() 应返回异步上下文管理器,让内存和 PostgreSQL 两种实现拥有统一的打开、关闭方式。不能只把 checkpointer 传给编译图后,再假设可以从图对象中取回并关闭它。
最后,把这个 lifespan 工厂交给应用工厂:
01from fastapi import FastAPI0203from app.api.router import router as api_router04from app.core.lifespan import create_lifespan05from app.core.settings import Settings, get_settings06from app.middleware import register_middleware0708AppLifespan = Callable[[FastAPI], AbstractAsyncContextManager[None]]091011def create_app(12settings: Settings | None = None,13*,14lifespan: AppLifespan | None = None,15) -> FastAPI:16current = settings if settings is not None else get_settings()17is_production = current.app.environment == "production"18current_lifespan = (19lifespan if lifespan is not None else create_lifespan(current)20)2122app = FastAPI(23title=current.app.name,24version="0.1.0",25debug=current.app.environment == "development",26openapi_url=None if is_production else "/openapi.json",27docs_url=None if is_production else "/docs",28redoc_url=None if is_production else "/redoc",29lifespan=current_lifespan,30)31app.state.settings = current3233register_middleware(app, current)34app.include_router(api_router)35return app
正常启动没有传入 lifespan 时,工厂会调用 create_lifespan(current);测试则可以传入不连接外部服务的生命周期。这样应用工厂、路由依赖、中间件和 lifespan 始终使用同一份 Settings。这里选择在生产环境关闭 OpenAPI Schema 和两个文档页面,但这属于产品策略,不是固定答案。即使关闭这些入口,接口本身仍需认证、授权和输入校验,不能把隐藏文档当作安全措施。
5. 资源依赖
请求需要使用生命周期资源时,可以通过依赖从 request.app.state 读取:
01from typing import Annotated0203from fastapi import Depends, Request04from langchain_core.language_models import BaseChatModel050607def get_chat_model(request: Request) -> BaseChatModel:08return request.app.state.model091011ChatModelDep = Annotated[BaseChatModel, Depends(get_chat_model)]
路由获得的是抽象模型接口:
01from fastapi import APIRouter02from pydantic import BaseModel, Field0304from app.api.dependencies import ChatModelDep0506router = APIRouter()070809class ChatRequest(BaseModel):10question: str = Field(min_length=1, max_length=4_000)111213class ChatResponse(BaseModel):14answer: str151617@router.post("/chat")18async def chat(payload: ChatRequest, model: ChatModelDep) -> ChatResponse:19response = await model.ainvoke(payload.question)20return ChatResponse(answer=str(response.content))
ChatRequest 继承 BaseModel,所以 FastAPI 会把 JSON 请求体解析到 payload,并在调用模型前检查问题是否为空、长度是否超出当前项目限制。如果直接把 question: str 写在函数参数中,它默认会被当作查询参数,而不是 JSON 请求体。
资源范围需要区分:
| 对象 | 典型范围 |
|---|---|
| Settings | 每进程一份 |
| 数据库 Engine/Pool | 每进程一份 |
| HTTP Client | 每进程复用 |
| LangChain 模型客户端 | 通常每进程复用 |
| LangGraph 编译图 | 通常每进程复用 |
| 数据库 Session | 每请求一份 |
| 当前用户身份 | 每请求解析 |
thread_id | 每个对话线程一个,由每次图调用显式携带 |
不能因为数据库 Engine 可以复用,就把一个可变的数据库 Session 也放到全局。thread_id 也不能设置成全进程共享值,否则不同用户的对话状态可能混在一起。配置决定如何创建资源,资源自身还要遵守正确生命周期。
6. 服务与测试
第 4 节的完整应用工厂已经根据 settings.app.environment 设置标题、调试模式和文档地址。debug 只在 development 环境开启,不能简单写成 not is_production,否则 test 和 staging 也会意外进入调试模式。
CORS 同样要把 Settings 中的值真正传给中间件:
01from fastapi import FastAPI02from fastapi.middleware.cors import CORSMiddleware0304from app.core.settings import Settings050607def add_cors(app: FastAPI, settings: Settings) -> None:08app.add_middleware(09CORSMiddleware,10allow_origins=settings.server.cors.allow_origins,11allow_methods=settings.server.cors.allow_methods,12allow_headers=["Authorization", "Content-Type"],13allow_credentials=False,14)
上一篇的 CorsSettings 只声明了来源和请求方法,因此这里为请求头和凭据使用明确的安全值。如果项目需要让这两项随环境变化,应先把 allow_headers、allow_credentials 同时加入 Settings 模型和 TOML,再从 settings.server.cors 读取。配置项只有传入真实组件才会生效,仅在 TOML 中写下 allow_origins 不会自动改变浏览器跨域行为。
测试与排查
应用工厂允许测试直接构造 Settings:
01from collections.abc import AsyncIterator02from contextlib import asynccontextmanager0304import pytest05from fastapi import FastAPI0607from app.core.settings import Settings08from app.factory import create_app091011@asynccontextmanager12async def empty_lifespan(_app: FastAPI) -> AsyncIterator[None]:13yield141516@pytest.fixture17def test_settings() -> Settings:18return Settings(19_env_file=None,20app={21"name": "FastAPI LangChain Test",22"environment": "test",23"log_level": "WARNING",24},25server={26"host": "127.0.0.1",27"port": 8000,28"cors": {29"allow_origins": ["http://testserver"],30"allow_methods": ["GET", "POST", "OPTIONS"],31},32},33model={34"provider": "fake",35"name": "fake-model",36"base_url": "https://example.test",37"api_key": None,38"temperature": 0,39"timeout_seconds": 5,40"max_retries": 0,41},42langgraph={43"recursion_limit": 10,44"checkpoint_backend": "memory",45},46database={47"url": None,48"pool_size": 1,49"pool_timeout_seconds": 5,50},51langsmith_tracing=False,52langsmith_api_key=None,53)545556@pytest.fixture57def app(test_settings: Settings) -> FastAPI:58return create_app(test_settings, lifespan=empty_lifespan)
_env_file=None 会关闭 .env 读取;各个必填分组也都显式提供测试值,因此不会只覆盖两个字段却意外沿用生产密钥。第 42 篇自定义的 TOML 来源仍会参与收集,但初始化参数优先级最高。若配置测试需要完全隔离 TOML,应为测试定义专用 Settings 子类或测试配置文件,这部分会在后续测试章节展开。
empty_lifespan 让只测试应用元数据或普通路由的用例不连接真实模型和数据库。需要验证资源型路由时,应改用能够写入 app.state.model 等假资源的测试 lifespan,或覆盖对应的 FastAPI 依赖。
创建 TestClient 时,如果要让 lifespan 真正运行,应把它作为上下文管理器使用。下面运行的是 fixture 中注入的 empty_lifespan,不会连接外部服务:
1from fastapi.testclient import TestClient234def test_runtime_info(app):5with TestClient(app) as client:6response = client.get("/runtime-info")78assert response.status_code == 200
如果测试必须修改环境变量,应在修改后清除缓存,并在测试结束时恢复环境。pytest 的 monkeypatch 可以管理这个过程:
01from app.core.settings import get_settings020304def test_model_override(monkeypatch):05monkeypatch.setenv("MODEL__NAME", "test-model")06get_settings.cache_clear()0708try:09settings = get_settings()10assert settings.model.name == "test-model"11finally:12get_settings.cache_clear()
monkeypatch 会在测试结束时恢复环境变量,finally 则保证即使断言失败也会清除 Settings 缓存。