创建时间: 2026-08-28最后更新: 2026-08-28

1. 入口

FastAPI 配置分为两层。pyproject.toml 中的 [tool.fastapi] 配置 CLI 如何找到应用;应用 Settings 则配置创建出来的服务如何运行。

当前项目使用:

pyproject.toml
1
[tool.fastapi]
2
entrypoint = "app.main:app"

冒号左侧是模块路径,右侧是模块中的对象名:

fastapi-entrypoint.txt
1
app.main:app
2
│ │ └── app/main.py 中的 FastAPI 对象
3
│ └────── app/main.py 模块
4
└────────── app Python 包

对应代码是:

app/main.py
1
from app.factory import create_app
2
3
app = create_app()

FastAPI CLI 执行的效果相当于:

import-entrypoint.py
1
from app.main import app

配置入口后,本地开发不必重复传文件路径:

fastapi-commands.bash
1
uv run fastapi dev
2
uv 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,入口也必须同步调整:

src-entrypoint.toml
1
[tool.fastapi]
2
entrypoint = "agent_api.main:app"

这个写法要求 agent_api 已经按照 src 布局安装到当前 Python 环境中。只是在仓库里创建 src/agent_api 目录,却没有正确配置并安装项目时,Python 仍然可能找不到这个包。

入口配置错误时,FastAPI CLI 会在启动阶段报告模块无法导入或找不到 app,它与请求路由返回 404 是不同问题。

2. 应用工厂

不要在 main.py 中读取几十个环境变量,再逐个创建中间件、数据库和模型。应用工厂可以把组装过程集中起来:

app/factory.py
01
from collections.abc import Callable
02
from contextlib import AbstractAsyncContextManager
03
04
from fastapi import FastAPI
05
06
from app.api.router import router as api_router
07
from app.core.settings import Settings, get_settings
08
from app.middleware import register_middleware
09
10
11
def create_app(settings: Settings | None = None) -> FastAPI:
12
current_settings = settings if settings is not None else get_settings()
13
14
app = FastAPI(
15
title=current_settings.app.name,
16
version="0.1.0",
17
debug=current_settings.app.environment == "development",
18
)
19
app.state.settings = current_settings
20
21
register_middleware(app, current_settings)
22
app.include_router(api_router)
23
return app

settings 参数允许测试显式传入配置;正常启动则调用 get_settings()。这比在测试中修改全局环境变量后重新导入整个模块更可控。

app.state.settings 把当前配置挂在应用实例上,适合需要从 Request 找到应用资源的依赖。它不是让所有业务代码直接访问全局状态的理由,业务服务仍应通过构造参数接收自己需要的配置。

Settings 可以在进程中缓存:

app/core/settings.py
1
from functools import lru_cache
2
3
4
@lru_cache(maxsize=1)
5
def get_settings() -> Settings:
6
return Settings()

缓存后,同一 worker 进程只读取和校验一次配置。部署多个 worker 时,每个进程都有自己的 Settings 和资源,Python 内存不会跨进程共享。

修改 .env 或 TOML 后,缓存对象不会自行变化。生产环境应重启或滚动更新应用,而不是尝试在请求中清除缓存。

3. 依赖

FastAPI 依赖可以把 Settings 提供给路由:

app/api/dependencies.py
01
from typing import Annotated
02
03
from fastapi import Depends, Request
04
05
from app.core.settings import Settings
06
07
08
def get_current_settings(request: Request) -> Settings:
09
return request.app.state.settings
10
11
12
SettingsDep = Annotated[Settings, Depends(get_current_settings)]

get_settings() 负责在创建应用时读取配置,get_current_settings() 则负责把这个已经确定的对象交给路由。名称看起来相近,但职责不同。这样测试向应用工厂传入 Settings 后,路由、中间件和 lifespan 都会看到同一个实例。

路由只声明需要配置,不负责读取配置文件:

app/api/routes/info.py
01
from fastapi import APIRouter
02
03
from app.api.dependencies import SettingsDep
04
05
router = APIRouter()
06
07
08
@router.get("/runtime-info")
09
async def runtime_info(settings: SettingsDep) -> dict[str, str]:
10
return {
11
"app_name": settings.app.name,
12
"environment": settings.app.environment,
13
"model": settings.model.name,
14
}

真实接口不一定应该公开模型名称和环境信息。这里用于演示依赖传递;生产健康检查只返回排障所需的非敏感字段。

依赖注入的价值不是少写一行导入,而是让对象来源明确且可以替换。对于 Settings,应用工厂参数就是最直接的替换入口:

tests/test_info.py
1
from app.factory import create_app
2
3
4
def test_factory_uses_injected_settings(test_settings):
5
app = create_app(test_settings)
6
7
assert app.state.settings is test_settings

这里的 test_settings fixture 会在第 6 节定义。模型客户端等资源依赖仍然可以使用 app.dependency_overrides 替换,但覆盖完成后应及时清空,避免影响其他测试。

如果路由自己调用 Settings(),测试就很难替换来源,还可能在每次请求中重复读取文件。

4. 生命周期

Settings 是轻量配置对象,数据库连接池、异步 HTTP 客户端和 LangGraph checkpointer 则是需要创建与关闭的长期资源。它们适合放进 FastAPI lifespan,也就是从应用启动到应用停止的这段生命周期:

app/core/lifespan.py
01
from collections.abc import AsyncIterator
02
from contextlib import asynccontextmanager
03
04
from fastapi import FastAPI
05
from httpx import AsyncClient, Timeout
06
07
from app.agent.graph import create_graph
08
from app.agent.model import create_model
09
from app.core.checkpointer import open_checkpointer
10
from app.core.settings import Settings
11
12
13
def create_lifespan(settings: Settings):
14
@asynccontextmanager
15
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
16
async with AsyncClient(
17
timeout=Timeout(settings.model.timeout_seconds),
18
) as http_client:
19
async with open_checkpointer(settings) as checkpointer:
20
model = create_model(settings)
21
graph = create_graph(
22
model=model,
23
checkpointer=checkpointer,
24
)
25
26
app.state.http_client = http_client
27
app.state.model = model
28
app.state.checkpointer = checkpointer
29
app.state.graph = graph
30
31
yield
32
33
return 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 工厂交给应用工厂:

app/factory.py
01
from fastapi import FastAPI
02
03
from app.api.router import router as api_router
04
from app.core.lifespan import create_lifespan
05
from app.core.settings import Settings, get_settings
06
from app.middleware import register_middleware
07
08
AppLifespan = Callable[[FastAPI], AbstractAsyncContextManager[None]]
09
10
11
def create_app(
12
settings: Settings | None = None,
13
*,
14
lifespan: AppLifespan | None = None,
15
) -> FastAPI:
16
current = settings if settings is not None else get_settings()
17
is_production = current.app.environment == "production"
18
current_lifespan = (
19
lifespan if lifespan is not None else create_lifespan(current)
20
)
21
22
app = FastAPI(
23
title=current.app.name,
24
version="0.1.0",
25
debug=current.app.environment == "development",
26
openapi_url=None if is_production else "/openapi.json",
27
docs_url=None if is_production else "/docs",
28
redoc_url=None if is_production else "/redoc",
29
lifespan=current_lifespan,
30
)
31
app.state.settings = current
32
33
register_middleware(app, current)
34
app.include_router(api_router)
35
return app

正常启动没有传入 lifespan 时,工厂会调用 create_lifespan(current);测试则可以传入不连接外部服务的生命周期。这样应用工厂、路由依赖、中间件和 lifespan 始终使用同一份 Settings。这里选择在生产环境关闭 OpenAPI Schema 和两个文档页面,但这属于产品策略,不是固定答案。即使关闭这些入口,接口本身仍需认证、授权和输入校验,不能把隐藏文档当作安全措施。

5. 资源依赖

请求需要使用生命周期资源时,可以通过依赖从 request.app.state 读取:

app/api/dependencies.py
01
from typing import Annotated
02
03
from fastapi import Depends, Request
04
from langchain_core.language_models import BaseChatModel
05
06
07
def get_chat_model(request: Request) -> BaseChatModel:
08
return request.app.state.model
09
10
11
ChatModelDep = Annotated[BaseChatModel, Depends(get_chat_model)]

路由获得的是抽象模型接口:

app/api/routes/chat.py
01
from fastapi import APIRouter
02
from pydantic import BaseModel, Field
03
04
from app.api.dependencies import ChatModelDep
05
06
router = APIRouter()
07
08
09
class ChatRequest(BaseModel):
10
question: str = Field(min_length=1, max_length=4_000)
11
12
13
class ChatResponse(BaseModel):
14
answer: str
15
16
17
@router.post("/chat")
18
async def chat(payload: ChatRequest, model: ChatModelDep) -> ChatResponse:
19
response = await model.ainvoke(payload.question)
20
return 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 中的值真正传给中间件:

app/middleware/cors.py
01
from fastapi import FastAPI
02
from fastapi.middleware.cors import CORSMiddleware
03
04
from app.core.settings import Settings
05
06
07
def add_cors(app: FastAPI, settings: Settings) -> None:
08
app.add_middleware(
09
CORSMiddleware,
10
allow_origins=settings.server.cors.allow_origins,
11
allow_methods=settings.server.cors.allow_methods,
12
allow_headers=["Authorization", "Content-Type"],
13
allow_credentials=False,
14
)

上一篇的 CorsSettings 只声明了来源和请求方法,因此这里为请求头和凭据使用明确的安全值。如果项目需要让这两项随环境变化,应先把 allow_headers、allow_credentials 同时加入 Settings 模型和 TOML,再从 settings.server.cors 读取。配置项只有传入真实组件才会生效,仅在 TOML 中写下 allow_origins 不会自动改变浏览器跨域行为。

测试与排查

应用工厂允许测试直接构造 Settings:

tests/conftest.py
01
from collections.abc import AsyncIterator
02
from contextlib import asynccontextmanager
03
04
import pytest
05
from fastapi import FastAPI
06
07
from app.core.settings import Settings
08
from app.factory import create_app
09
10
11
@asynccontextmanager
12
async def empty_lifespan(_app: FastAPI) -> AsyncIterator[None]:
13
yield
14
15
16
@pytest.fixture
17
def test_settings() -> Settings:
18
return Settings(
19
_env_file=None,
20
app={
21
"name": "FastAPI LangChain Test",
22
"environment": "test",
23
"log_level": "WARNING",
24
},
25
server={
26
"host": "127.0.0.1",
27
"port": 8000,
28
"cors": {
29
"allow_origins": ["http://testserver"],
30
"allow_methods": ["GET", "POST", "OPTIONS"],
31
},
32
},
33
model={
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
},
42
langgraph={
43
"recursion_limit": 10,
44
"checkpoint_backend": "memory",
45
},
46
database={
47
"url": None,
48
"pool_size": 1,
49
"pool_timeout_seconds": 5,
50
},
51
langsmith_tracing=False,
52
langsmith_api_key=None,
53
)
54
55
56
@pytest.fixture
57
def app(test_settings: Settings) -> FastAPI:
58
return 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,不会连接外部服务:

test-client-lifespan.py
1
from fastapi.testclient import TestClient
2
3
4
def test_runtime_info(app):
5
with TestClient(app) as client:
6
response = client.get("/runtime-info")
7
8
assert response.status_code == 200

如果测试必须修改环境变量,应在修改后清除缓存,并在测试结束时恢复环境。pytest 的 monkeypatch 可以管理这个过程:

tests/test_settings.py
01
from app.core.settings import get_settings
02
03
04
def test_model_override(monkeypatch):
05
monkeypatch.setenv("MODEL__NAME", "test-model")
06
get_settings.cache_clear()
07
08
try:
09
settings = get_settings()
10
assert settings.model.name == "test-model"
11
finally:
12
get_settings.cache_clear()

monkeypatch 会在测试结束时恢复环境变量,finally 则保证即使断言失败也会清除 Settings 缓存。

正在验证登录状态
请稍候,验证完成后将继续显示文章内容