拓冰网站定制 
系列导航上篇核心机制、项目配置与完整请求链路下篇数据库、JWT、测试部署与热门项目架构拆解本文摘要会写路由不等于会搭建可靠后端。下篇先把 FastAPI 放进完整工程讲清配置、数据库、事务、JWT、测试和部署再进入 Open WebUI、Langflow 与 Full Stack FastAPI Template 的核心源码观察同一套原则如何在真实项目中落地。一、先确定工程边界一个中等规模 FastAPI 项目可以采用以下结构app/ ├── main.py # 创建应用、注册路由和中间件 ├── api/ │ ├── deps.py # 数据库、当前用户等依赖 │ └── routes/ # HTTP 层 ├── core/ │ ├── config.py # 环境配置 │ └── security.py # 密码与 Token ├── db/ │ ├── session.py # Engine 和 Session │ └── models.py # ORM 模型 ├── schemas/ # 请求与响应模型 ├── repositories/ # 数据访问 ├── services/ # 业务规则 └── tests/推荐的调用方向是Router → Service → Repository → DatabaseRouter 负责 HTTP 协议Service 负责业务规则Repository 负责数据访问。不要让路由函数同时完成参数校验、SQL、事务、邮件发送和权限判断否则测试与复用都会迅速变难。二、配置管理不要在代码里散落环境变量安装设置组件pip install pydantic-settingsapp/core/config.pyfromfunctoolsimportlru_cachefrompydanticimportSecretStrfrompydantic_settingsimportBaseSettings,SettingsConfigDictclassSettings(BaseSettings):app_name:strFastAPI Serviceenvironment:strdevelopmentdatabase_url:strsqlite:///./app.dbjwt_secret:SecretStr access_token_minutes:int30cors_origins:list[str][http://localhost:5173]model_configSettingsConfigDict(env_file.env,env_file_encodingutf-8,case_sensitiveFalse,extraignore,)lru_cachedefget_settings()-Settings:returnSettings().envAPP_NAMEFastAPI Demo ENVIRONMENTdevelopment DATABASE_URLpostgresqlpsycopg://app:passwordlocalhost:5432/app JWT_SECRET请替换为随机长字符串 ACCESS_TOKEN_MINUTES30 CORS_ORIGINS[http://localhost:5173]SecretStr能降低日志或调试输出意外展示密钥的风险但它不是加密存储。生产密钥仍应放在云 Secret Manager、Kubernetes Secret 或部署平台的安全变量中。lru_cache让配置对象在进程内只创建一次。测试中如果修改了环境变量应调用get_settings.cache_clear()。三、数据库会话与请求生命周期为了让示例保持清晰下面使用 SQLModel。它建立在 SQLAlchemy 与 Pydantic 之上pip install sqlmodel psycopg[binary]app/db/session.pyfromcollections.abcimportGeneratorfromsqlmodelimportSession,create_enginefromapp.core.configimportget_settings settingsget_settings()enginecreate_engine(settings.database_url,pool_pre_pingTrue,)defget_session()-Generator[Session,None,None]:withSession(engine)assession:yieldsession这里的yield与请求生命周期绑定进入路径函数前创建 Session请求结束后关闭。依赖函数本身不应该自动 commit因为一个业务操作可能包含多次写入事务边界更适合放在 Service 层。模型与 SchemafromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDictfromsqlmodelimportField,SQLModelclassTask(SQLModel,tableTrue):id:int|NoneField(defaultNone,primary_keyTrue)title:strField(indexTrue,max_length200)completed:boolFalsecreated_at:datetimeField(default_factorydatetime.utcnow)classTaskCreate(BaseModel):title:strclassTaskPublic(BaseModel):model_configConfigDict(from_attributesTrue)id:inttitle:strcompleted:boolcreated_at:datetimeRepositoryfromsqlmodelimportSession,selectfromapp.db.modelsimportTaskclassTaskRepository:def__init__(self,session:Session):self.sessionsessiondeflist(self)-list[Task]:returnlist(self.session.exec(select(Task)).all())defadd(self,task:Task)-Task:self.session.add(task)self.session.flush()self.session.refresh(task)returntaskService 决定事务classTaskService:def__init__(self,session:Session):self.sessionsession self.repoTaskRepository(session)defcreate(self,title:str)-Task:ifnottitle.strip():raiseValueError(title cannot be empty)try:taskself.repo.add(Task(titletitle.strip()))self.session.commit()returntaskexceptException:self.session.rollback()raise路由只负责协议转换fromtypingimportAnnotatedfromfastapiimportAPIRouter,Depends,statusfromsqlmodelimportSession routerAPIRouter(prefix/tasks,tags[tasks])SessionDepAnnotated[Session,Depends(get_session)]router.post(,response_modelTaskPublic,status_codestatus.HTTP_201_CREATED)defcreate_task(payload:TaskCreate,session:SessionDep):returnTaskService(session).create(payload.title)同步 SQLAlchemy/SQLModel 应配合同步def路由FastAPI 会将其放入线程池。如果选择 SQLAlchemy AsyncSession则应从驱动、Session 到 Repository 全链路异步不要混用。四、认证OAuth2PasswordBearer 只是取 TokenOAuth2PasswordBearer不会自动验证用户它主要完成两件事告诉 OpenAPI 使用 Bearer Token并从 Authorization Header 提取 Token。fromtypingimportAnnotatedfromfastapiimportDependsfromfastapi.securityimportOAuth2PasswordBearer oauth2_schemeOAuth2PasswordBearer(tokenUrl/api/v1/auth/token)TokenDepAnnotated[str,Depends(oauth2_scheme)]安装密码和 JWT 工具pip installpwdlib[argon2]pyjwtapp/core/security.pyfromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompwdlibimportPasswordHashfromapp.core.configimportget_settings password_hashPasswordHash.recommended()defhash_password(password:str)-str:returnpassword_hash.hash(password)defverify_password(password:str,hashed:str)-bool:returnpassword_hash.verify(password,hashed)defcreate_access_token(subject:str)-str:settingsget_settings()nowdatetime.now(timezone.utc)payload{sub:subject,iat:now,exp:nowtimedelta(minutessettings.access_token_minutes),}returnjwt.encode(payload,settings.jwt_secret.get_secret_value(),algorithmHS256,)解析当前用户fromfastapiimportHTTPException,statusfromjwtimportInvalidTokenErrorasyncdefget_current_user(token:TokenDep,session:SessionDep):credentials_errorHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailCould not validate credentials,headers{WWW-Authenticate:Bearer},)try:payloadjwt.decode(token,get_settings().jwt_secret.get_secret_value(),algorithms[HS256],)user_idint(payload[sub])except(InvalidTokenError,KeyError,ValueError):raisecredentials_error usersession.get(User,user_id)ifuserisNoneornotuser.is_active:raisecredentials_errorreturnuser权限可以继续建立在当前用户依赖上defrequire_admin(user:CurrentUser):ifnotuser.is_admin:raiseHTTPException(status_code403,detailAdmin required)returnuser认证是 401已登录但权限不足是 403两者不要混用。五、中间件与 CORSfromfastapi.middleware.corsimportCORSMiddleware settingsget_settings()app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)如果allow_credentialsTrue生产环境不要简单使用allow_origins[*]应明确列出可信前端来源。自定义请求耗时中间件importtimeimportuuidfromfastapiimportRequestapp.middleware(http)asyncdefrequest_context(request:Request,call_next):request_idrequest.headers.get(X-Request-ID,str(uuid.uuid4()))startedtime.perf_counter()responseawaitcall_next(request)response.headers[X-Request-ID]request_id response.headers[X-Process-Time]str(round(time.perf_counter()-started,6))returnresponse中间件适合跨接口的日志、追踪 ID、安全 Header 和性能统计不适合塞入具体业务判断。六、后台任务适合轻任务不是任务队列fromfastapiimportBackgroundTasksdefwrite_audit_log(task_id:int)-None:withopen(audit.log,a,encodingutf-8)asfile:file.write(fcreated task{task_id}\n)router.post(,response_modelTaskPublic)defcreate_task(payload:TaskCreate,session:SessionDep,background_tasks:BackgroundTasks,):taskTaskService(session).create(payload.title)background_tasks.add_task(write_audit_log,task.id)returntaskBackgroundTasks 会在响应发送后、当前应用进程中执行。它适合短小、失败可容忍的工作。视频转码、大批量邮件、模型推理等长任务应使用 Celery、Dramatiq、RQ 或独立消息队列因为进程重启会丢失内存中的后台任务。七、Lifespan初始化和释放共享资源fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIasynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.http_clientAsyncClient(timeout10)app.state.modelawaitload_model()yieldawaitapp.state.http_client.aclose()awaitapp.state.model.close()appFastAPI(lifespanlifespan)yield前只执行一次启动逻辑yield后执行关闭逻辑。数据库连接池、HTTP Client、模型和缓存客户端适合在这里管理。不要在 import 模块时执行昂贵初始化否则测试收集、CLI 工具和多 Worker 启动都会受到影响。八、测试覆盖依赖边界而不是启动真实服务器FastAPI 的 TestClient 基于 HTTPXfromfastapi.testclientimportTestClientfromapp.mainimportapp clientTestClient(app)deftest_create_task():responseclient.post(/api/v1/tasks,json{title:write tests})assertresponse.status_code201assertresponse.json()[title]write tests认证依赖可以替换deffake_current_user():returnUser(id1,emailtesterexample.com,is_activeTrue)app.dependency_overrides[get_current_user]fake_current_userdeftest_private_endpoint():responseclient.get(/api/v1/profile)assertresponse.status_code200defteardown_module():app.dependency_overrides.clear()依赖覆盖比在测试中签发真实 JWT 更适合路由单元测试认证编解码本身再使用独立测试覆盖。异步测试可以使用 HTTPX AsyncClient 和 ASGITransportimportpytestfromhttpximportASGITransport,AsyncClientpytest.mark.anyioasyncdeftest_root():transportASGITransport(appapp)asyncwithAsyncClient(transporttransport,base_urlhttp://test,)asclient:responseawaitclient.get(/)assertresponse.status_code200九、生产部署需要考虑什么本地开发fastapi dev app/main.py单进程生产启动fastapi run app/main.py--host 0.0.0.0--port 8000使用 Uvicornuvicorn app.main:app--host0.0.0.0--port8000--workers4Worker 数不是越多越好。每个 Worker 都是独立进程会分别创建连接池、缓存和模型对象应结合 CPU、内存与压测结果决定。一个基础 DockerfileFROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app CMD [ fastapi, run, app/main.py, --host, 0.0.0.0, --port, 8000 ]生产环境还需要反向代理或云负载均衡HTTPS 和可信代理 Header 配置数据库迁移而不是启动时自动建表readiness 与 liveness 检查结构化日志、指标和分布式追踪请求超时、限流和上传大小限制优雅关闭与滚动发布独立的后台任务系统。十、把应用装配集中到 main.pyfromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromapp.api.routesimportauth,tasks,usersfromapp.core.configimportget_settingsfromapp.lifecycleimportlifespandefcreate_app()-FastAPI:settingsget_settings()appFastAPI(titlesettings.app_name,lifespanlifespan,)app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)app.include_router(auth.router,prefix/api/v1)app.include_router(users.router,prefix/api/v1)app.include_router(tasks.router,prefix/api/v1)returnapp appcreate_app()Application Factory 让测试可以在不同配置下创建应用也把路由、中间件和生命周期的装配位置固定下来。十一、从热门项目反推生产级架构前面的工程结构不是凭空设计出来的。下面选择三个采用 FastAPI 的代表性仓库重点看它们的应用创建、生命周期、路由组织和数据边界而不是复述 README。1. Open WebUI让 FastAPI 成为 AI 平台的统一入口Open WebUI 的后端不仅提供普通 REST API还要协调模型连接、认证、知识库、文件、WebSocket 和前端静态资源。它在创建应用时显式传入lifespan并根据环境决定是否暴露文档appFastAPI(docs_url/docsifENVdevelseNone,openapi_url/openapi.jsonifENVdevelseNone,lifespanlifespan,)这种写法包含两个重要判断生产环境不必默认暴露交互式文档连接池、模型客户端和缓存等共享资源应该交给生命周期统一创建和释放而不是散落在模块导入阶段。Open WebUI 还大量使用app.state保存进程级共享状态。它适合保存配置快照和连接客户端但不适合保存请求级用户数据。请求身份仍应通过依赖注入传递否则并发请求之间容易相互污染。路由层按用户、模型、文件、知识库等领域拆分再由主应用统一挂载app.include_router(users.router,prefix/api/v1/users)app.include_router(models.router,prefix/api/v1/models)app.include_router(files.router,prefix/api/v1/files)可借鉴的不是路由数量而是“领域模块拥有自己的入口主程序只负责装配”。新增业务时变化被限制在对应模块内。2. LangflowApplication Factory 与复杂 LifespanLangflow 需要加载组件、数据库、缓存、可观测性和执行引擎。它采用创建函数组织应用使测试、命令行和部署入口可以用不同配置得到 FastAPI 实例defcreate_app()-FastAPI:appFastAPI(lifespanlifespan)app.include_router(api_router,prefix/api/v1)register_exception_handlers(app)configure_cors(app)returnapp复杂启动逻辑被放进lifespan并使用yield分隔初始化与清理asynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.servicesawaitbuild_services()awaitapp.state.services.start()try:yieldfinally:awaitapp.state.services.stop()关键点是清理代码位于finally即使运行期间出现异常服务也有机会关闭连接。对需要热重载或多 Worker 的项目还要保证初始化具有幂等性不能假设它只执行一次。Langflow 的 HTTP 路由只处理协议问题Service 负责业务编排执行引擎负责流程运行。路由不直接控制复杂组件避免 Web 层逐渐变成难以测试的“大函数”。3. Full Stack FastAPI Template最适合照着练习的工程样板官方模板把 FastAPI、SQLModel、PostgreSQL、JWT、React、Pytest、Playwright 和 Docker Compose 放在一个完整项目中。它尤其值得学习三点。第一ORM 实体与公开 Schema 分工。数据库模型可以包含内部字段接口响应模型只声明允许返回的内容classUserBase(SQLModel):email:EmailStrField(uniqueTrue,indexTrue)is_active:boolTrueclassUserCreate(UserBase):password:strField(min_length8,max_length128)classUserPublic(UserBase):id:UUID第二Session 和当前用户通过依赖传入路由不自行创建数据库连接SessionDepAnnotated[Session,Depends(get_db)]CurrentUserAnnotated[User,Depends(get_current_user)]router.get(/me,response_modelUserPublic)defread_me(current_user:CurrentUser):returncurrent_user第三前端客户端可以根据 OpenAPI 生成。后端的类型和响应模型因此不只是文档也成为前后端协作契约。随意返回未声明字段会让生成客户端和实际响应逐渐偏离。十二、三个项目的模式对照项目FastAPI 承担的角色最值得学习的模式更适合的场景Open WebUIAI 平台统一后端大量领域路由、共享状态、统一异常AI 对话、模型网关、知识库Langflow可视化执行平台 APIApplication Factory、复杂 Lifespan、Service 编排工作流、插件系统、执行引擎Full Stack FastAPI Template全栈业务后端Schema 边界、Session 依赖、JWT、测试与容器管理后台、SaaS、标准 CRUD三者规模不同却共享同一条演进路径薄路由、显式依赖、清晰数据边界、集中装配、生命周期管理和自动化测试。新项目不需要复制任何一个仓库的全部目录而应按实际复杂度逐步引入这些边界。十三、推荐的渐进式项目结构app/ ├── main.py # create_app、路由和中间件装配 ├── api/ │ ├── deps.py # Session、当前用户、权限 │ └── routes/ # 按领域组织 HTTP 接口 ├── core/ # 配置、安全、日志 ├── db/ # Engine、Session、ORM 模型 ├── schemas/ # 输入与公开响应模型 ├── repositories/ # 查询和持久化 ├── services/ # 业务编排与事务边界 └── tests/ # 单元、接口和集成测试小项目可以先保留api core db。只有当路由开始重复查询或业务规则时再提取 Repository 和 Service。分层的目的不是增加文件数量而是让每个边界能够被独立理解和测试。总结FastAPI 工程化的核心不是堆组件而是明确生命周期和边界配置对象负责从环境加载配置依赖管理请求级资源Service 控制业务与事务Repository 隔离数据访问认证依赖解析当前身份中间件处理跨接口能力Lifespan 管理进程级资源测试通过依赖覆盖隔离外部系统部署层解决多进程、代理、监控和故障恢复。从 Open WebUI、Langflow 和官方模板可以看到FastAPI 的价值不仅是快速写出接口更在于它允许项目从类型声明和依赖注入开始平滑演进到具备数据库、认证、测试、可观测性与复杂生命周期的生产系统。返回上篇FastAPI 实战教程上核心机制、项目配置与完整请求链路参考资料SQL Databaseshttps://fastapi.tiangolo.com/tutorial/sql-databases/Securityhttps://fastapi.tiangolo.com/tutorial/security/Middlewarehttps://fastapi.tiangolo.com/tutorial/middleware/Background Taskshttps://fastapi.tiangolo.com/tutorial/background-tasks/Lifespanhttps://fastapi.tiangolo.com/advanced/events/Testinghttps://fastapi.tiangolo.com/tutorial/testing/Deployment Conceptshttps://fastapi.tiangolo.com/deployment/concepts/Open WebUIhttps://github.com/open-webui/open-webuiLangflowhttps://github.com/langflow-ai/langflowFull Stack FastAPI Templatehttps://github.com/fastapi/full-stack-fastapi-template FastAPI 实战教程（下）：数据库、JWT、测试部署与热门项目架构拆解
系列导航上篇核心机制、项目配置与完整请求链路下篇数据库、JWT、测试部署与热门项目架构拆解本文摘要会写路由不等于会搭建可靠后端。下篇先把 FastAPI 放进完整工程讲清配置、数据库、事务、JWT、测试和部署再进入 Open WebUI、Langflow 与 Full Stack FastAPI Template 的核心源码观察同一套原则如何在真实项目中落地。一、先确定工程边界一个中等规模 FastAPI 项目可以采用以下结构app/ ├── main.py # 创建应用、注册路由和中间件 ├── api/ │ ├── deps.py # 数据库、当前用户等依赖 │ └── routes/ # HTTP 层 ├── core/ │ ├── config.py # 环境配置 │ └── security.py # 密码与 Token ├── db/ │ ├── session.py # Engine 和 Session │ └── models.py # ORM 模型 ├── schemas/ # 请求与响应模型 ├── repositories/ # 数据访问 ├── services/ # 业务规则 └── tests/推荐的调用方向是Router → Service → Repository → DatabaseRouter 负责 HTTP 协议Service 负责业务规则Repository 负责数据访问。不要让路由函数同时完成参数校验、SQL、事务、邮件发送和权限判断否则测试与复用都会迅速变难。二、配置管理不要在代码里散落环境变量安装设置组件pip install pydantic-settingsapp/core/config.pyfromfunctoolsimportlru_cachefrompydanticimportSecretStrfrompydantic_settingsimportBaseSettings,SettingsConfigDictclassSettings(BaseSettings):app_name:strFastAPI Serviceenvironment:strdevelopmentdatabase_url:strsqlite:///./app.dbjwt_secret:SecretStr access_token_minutes:int30cors_origins:list[str][http://localhost:5173]model_configSettingsConfigDict(env_file.env,env_file_encodingutf-8,case_sensitiveFalse,extraignore,)lru_cachedefget_settings()-Settings:returnSettings().envAPP_NAMEFastAPI Demo ENVIRONMENTdevelopment DATABASE_URLpostgresqlpsycopg://app:passwordlocalhost:5432/app JWT_SECRET请替换为随机长字符串 ACCESS_TOKEN_MINUTES30 CORS_ORIGINS[http://localhost:5173]SecretStr能降低日志或调试输出意外展示密钥的风险但它不是加密存储。生产密钥仍应放在云 Secret Manager、Kubernetes Secret 或部署平台的安全变量中。lru_cache让配置对象在进程内只创建一次。测试中如果修改了环境变量应调用get_settings.cache_clear()。三、数据库会话与请求生命周期为了让示例保持清晰下面使用 SQLModel。它建立在 SQLAlchemy 与 Pydantic 之上pip install sqlmodel psycopg[binary]app/db/session.pyfromcollections.abcimportGeneratorfromsqlmodelimportSession,create_enginefromapp.core.configimportget_settings settingsget_settings()enginecreate_engine(settings.database_url,pool_pre_pingTrue,)defget_session()-Generator[Session,None,None]:withSession(engine)assession:yieldsession这里的yield与请求生命周期绑定进入路径函数前创建 Session请求结束后关闭。依赖函数本身不应该自动 commit因为一个业务操作可能包含多次写入事务边界更适合放在 Service 层。模型与 SchemafromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDictfromsqlmodelimportField,SQLModelclassTask(SQLModel,tableTrue):id:int|NoneField(defaultNone,primary_keyTrue)title:strField(indexTrue,max_length200)completed:boolFalsecreated_at:datetimeField(default_factorydatetime.utcnow)classTaskCreate(BaseModel):title:strclassTaskPublic(BaseModel):model_configConfigDict(from_attributesTrue)id:inttitle:strcompleted:boolcreated_at:datetimeRepositoryfromsqlmodelimportSession,selectfromapp.db.modelsimportTaskclassTaskRepository:def__init__(self,session:Session):self.sessionsessiondeflist(self)-list[Task]:returnlist(self.session.exec(select(Task)).all())defadd(self,task:Task)-Task:self.session.add(task)self.session.flush()self.session.refresh(task)returntaskService 决定事务classTaskService:def__init__(self,session:Session):self.sessionsession self.repoTaskRepository(session)defcreate(self,title:str)-Task:ifnottitle.strip():raiseValueError(title cannot be empty)try:taskself.repo.add(Task(titletitle.strip()))self.session.commit()returntaskexceptException:self.session.rollback()raise路由只负责协议转换fromtypingimportAnnotatedfromfastapiimportAPIRouter,Depends,statusfromsqlmodelimportSession routerAPIRouter(prefix/tasks,tags[tasks])SessionDepAnnotated[Session,Depends(get_session)]router.post(,response_modelTaskPublic,status_codestatus.HTTP_201_CREATED)defcreate_task(payload:TaskCreate,session:SessionDep):returnTaskService(session).create(payload.title)同步 SQLAlchemy/SQLModel 应配合同步def路由FastAPI 会将其放入线程池。如果选择 SQLAlchemy AsyncSession则应从驱动、Session 到 Repository 全链路异步不要混用。四、认证OAuth2PasswordBearer 只是取 TokenOAuth2PasswordBearer不会自动验证用户它主要完成两件事告诉 OpenAPI 使用 Bearer Token并从 Authorization Header 提取 Token。fromtypingimportAnnotatedfromfastapiimportDependsfromfastapi.securityimportOAuth2PasswordBearer oauth2_schemeOAuth2PasswordBearer(tokenUrl/api/v1/auth/token)TokenDepAnnotated[str,Depends(oauth2_scheme)]安装密码和 JWT 工具pip installpwdlib[argon2]pyjwtapp/core/security.pyfromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompwdlibimportPasswordHashfromapp.core.configimportget_settings password_hashPasswordHash.recommended()defhash_password(password:str)-str:returnpassword_hash.hash(password)defverify_password(password:str,hashed:str)-bool:returnpassword_hash.verify(password,hashed)defcreate_access_token(subject:str)-str:settingsget_settings()nowdatetime.now(timezone.utc)payload{sub:subject,iat:now,exp:nowtimedelta(minutessettings.access_token_minutes),}returnjwt.encode(payload,settings.jwt_secret.get_secret_value(),algorithmHS256,)解析当前用户fromfastapiimportHTTPException,statusfromjwtimportInvalidTokenErrorasyncdefget_current_user(token:TokenDep,session:SessionDep):credentials_errorHTTPException(status_codestatus.HTTP_401_UNAUTHORIZED,detailCould not validate credentials,headers{WWW-Authenticate:Bearer},)try:payloadjwt.decode(token,get_settings().jwt_secret.get_secret_value(),algorithms[HS256],)user_idint(payload[sub])except(InvalidTokenError,KeyError,ValueError):raisecredentials_error usersession.get(User,user_id)ifuserisNoneornotuser.is_active:raisecredentials_errorreturnuser权限可以继续建立在当前用户依赖上defrequire_admin(user:CurrentUser):ifnotuser.is_admin:raiseHTTPException(status_code403,detailAdmin required)returnuser认证是 401已登录但权限不足是 403两者不要混用。五、中间件与 CORSfromfastapi.middleware.corsimportCORSMiddleware settingsget_settings()app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)如果allow_credentialsTrue生产环境不要简单使用allow_origins[*]应明确列出可信前端来源。自定义请求耗时中间件importtimeimportuuidfromfastapiimportRequestapp.middleware(http)asyncdefrequest_context(request:Request,call_next):request_idrequest.headers.get(X-Request-ID,str(uuid.uuid4()))startedtime.perf_counter()responseawaitcall_next(request)response.headers[X-Request-ID]request_id response.headers[X-Process-Time]str(round(time.perf_counter()-started,6))returnresponse中间件适合跨接口的日志、追踪 ID、安全 Header 和性能统计不适合塞入具体业务判断。六、后台任务适合轻任务不是任务队列fromfastapiimportBackgroundTasksdefwrite_audit_log(task_id:int)-None:withopen(audit.log,a,encodingutf-8)asfile:file.write(fcreated task{task_id}\n)router.post(,response_modelTaskPublic)defcreate_task(payload:TaskCreate,session:SessionDep,background_tasks:BackgroundTasks,):taskTaskService(session).create(payload.title)background_tasks.add_task(write_audit_log,task.id)returntaskBackgroundTasks 会在响应发送后、当前应用进程中执行。它适合短小、失败可容忍的工作。视频转码、大批量邮件、模型推理等长任务应使用 Celery、Dramatiq、RQ 或独立消息队列因为进程重启会丢失内存中的后台任务。七、Lifespan初始化和释放共享资源fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPIasynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.http_clientAsyncClient(timeout10)app.state.modelawaitload_model()yieldawaitapp.state.http_client.aclose()awaitapp.state.model.close()appFastAPI(lifespanlifespan)yield前只执行一次启动逻辑yield后执行关闭逻辑。数据库连接池、HTTP Client、模型和缓存客户端适合在这里管理。不要在 import 模块时执行昂贵初始化否则测试收集、CLI 工具和多 Worker 启动都会受到影响。八、测试覆盖依赖边界而不是启动真实服务器FastAPI 的 TestClient 基于 HTTPXfromfastapi.testclientimportTestClientfromapp.mainimportapp clientTestClient(app)deftest_create_task():responseclient.post(/api/v1/tasks,json{title:write tests})assertresponse.status_code201assertresponse.json()[title]write tests认证依赖可以替换deffake_current_user():returnUser(id1,emailtesterexample.com,is_activeTrue)app.dependency_overrides[get_current_user]fake_current_userdeftest_private_endpoint():responseclient.get(/api/v1/profile)assertresponse.status_code200defteardown_module():app.dependency_overrides.clear()依赖覆盖比在测试中签发真实 JWT 更适合路由单元测试认证编解码本身再使用独立测试覆盖。异步测试可以使用 HTTPX AsyncClient 和 ASGITransportimportpytestfromhttpximportASGITransport,AsyncClientpytest.mark.anyioasyncdeftest_root():transportASGITransport(appapp)asyncwithAsyncClient(transporttransport,base_urlhttp://test,)asclient:responseawaitclient.get(/)assertresponse.status_code200九、生产部署需要考虑什么本地开发fastapi dev app/main.py单进程生产启动fastapi run app/main.py--host 0.0.0.0--port 8000使用 Uvicornuvicorn app.main:app--host0.0.0.0--port8000--workers4Worker 数不是越多越好。每个 Worker 都是独立进程会分别创建连接池、缓存和模型对象应结合 CPU、内存与压测结果决定。一个基础 DockerfileFROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app CMD [ fastapi, run, app/main.py, --host, 0.0.0.0, --port, 8000 ]生产环境还需要反向代理或云负载均衡HTTPS 和可信代理 Header 配置数据库迁移而不是启动时自动建表readiness 与 liveness 检查结构化日志、指标和分布式追踪请求超时、限流和上传大小限制优雅关闭与滚动发布独立的后台任务系统。十、把应用装配集中到 main.pyfromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromapp.api.routesimportauth,tasks,usersfromapp.core.configimportget_settingsfromapp.lifecycleimportlifespandefcreate_app()-FastAPI:settingsget_settings()appFastAPI(titlesettings.app_name,lifespanlifespan,)app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)app.include_router(auth.router,prefix/api/v1)app.include_router(users.router,prefix/api/v1)app.include_router(tasks.router,prefix/api/v1)returnapp appcreate_app()Application Factory 让测试可以在不同配置下创建应用也把路由、中间件和生命周期的装配位置固定下来。十一、从热门项目反推生产级架构前面的工程结构不是凭空设计出来的。下面选择三个采用 FastAPI 的代表性仓库重点看它们的应用创建、生命周期、路由组织和数据边界而不是复述 README。1. Open WebUI让 FastAPI 成为 AI 平台的统一入口Open WebUI 的后端不仅提供普通 REST API还要协调模型连接、认证、知识库、文件、WebSocket 和前端静态资源。它在创建应用时显式传入lifespan并根据环境决定是否暴露文档appFastAPI(docs_url/docsifENVdevelseNone,openapi_url/openapi.jsonifENVdevelseNone,lifespanlifespan,)这种写法包含两个重要判断生产环境不必默认暴露交互式文档连接池、模型客户端和缓存等共享资源应该交给生命周期统一创建和释放而不是散落在模块导入阶段。Open WebUI 还大量使用app.state保存进程级共享状态。它适合保存配置快照和连接客户端但不适合保存请求级用户数据。请求身份仍应通过依赖注入传递否则并发请求之间容易相互污染。路由层按用户、模型、文件、知识库等领域拆分再由主应用统一挂载app.include_router(users.router,prefix/api/v1/users)app.include_router(models.router,prefix/api/v1/models)app.include_router(files.router,prefix/api/v1/files)可借鉴的不是路由数量而是“领域模块拥有自己的入口主程序只负责装配”。新增业务时变化被限制在对应模块内。2. LangflowApplication Factory 与复杂 LifespanLangflow 需要加载组件、数据库、缓存、可观测性和执行引擎。它采用创建函数组织应用使测试、命令行和部署入口可以用不同配置得到 FastAPI 实例defcreate_app()-FastAPI:appFastAPI(lifespanlifespan)app.include_router(api_router,prefix/api/v1)register_exception_handlers(app)configure_cors(app)returnapp复杂启动逻辑被放进lifespan并使用yield分隔初始化与清理asynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.servicesawaitbuild_services()awaitapp.state.services.start()try:yieldfinally:awaitapp.state.services.stop()关键点是清理代码位于finally即使运行期间出现异常服务也有机会关闭连接。对需要热重载或多 Worker 的项目还要保证初始化具有幂等性不能假设它只执行一次。Langflow 的 HTTP 路由只处理协议问题Service 负责业务编排执行引擎负责流程运行。路由不直接控制复杂组件避免 Web 层逐渐变成难以测试的“大函数”。3. Full Stack FastAPI Template最适合照着练习的工程样板官方模板把 FastAPI、SQLModel、PostgreSQL、JWT、React、Pytest、Playwright 和 Docker Compose 放在一个完整项目中。它尤其值得学习三点。第一ORM 实体与公开 Schema 分工。数据库模型可以包含内部字段接口响应模型只声明允许返回的内容classUserBase(SQLModel):email:EmailStrField(uniqueTrue,indexTrue)is_active:boolTrueclassUserCreate(UserBase):password:strField(min_length8,max_length128)classUserPublic(UserBase):id:UUID第二Session 和当前用户通过依赖传入路由不自行创建数据库连接SessionDepAnnotated[Session,Depends(get_db)]CurrentUserAnnotated[User,Depends(get_current_user)]router.get(/me,response_modelUserPublic)defread_me(current_user:CurrentUser):returncurrent_user第三前端客户端可以根据 OpenAPI 生成。后端的类型和响应模型因此不只是文档也成为前后端协作契约。随意返回未声明字段会让生成客户端和实际响应逐渐偏离。十二、三个项目的模式对照项目FastAPI 承担的角色最值得学习的模式更适合的场景Open WebUIAI 平台统一后端大量领域路由、共享状态、统一异常AI 对话、模型网关、知识库Langflow可视化执行平台 APIApplication Factory、复杂 Lifespan、Service 编排工作流、插件系统、执行引擎Full Stack FastAPI Template全栈业务后端Schema 边界、Session 依赖、JWT、测试与容器管理后台、SaaS、标准 CRUD三者规模不同却共享同一条演进路径薄路由、显式依赖、清晰数据边界、集中装配、生命周期管理和自动化测试。新项目不需要复制任何一个仓库的全部目录而应按实际复杂度逐步引入这些边界。十三、推荐的渐进式项目结构app/ ├── main.py # create_app、路由和中间件装配 ├── api/ │ ├── deps.py # Session、当前用户、权限 │ └── routes/ # 按领域组织 HTTP 接口 ├── core/ # 配置、安全、日志 ├── db/ # Engine、Session、ORM 模型 ├── schemas/ # 输入与公开响应模型 ├── repositories/ # 查询和持久化 ├── services/ # 业务编排与事务边界 └── tests/ # 单元、接口和集成测试小项目可以先保留api core db。只有当路由开始重复查询或业务规则时再提取 Repository 和 Service。分层的目的不是增加文件数量而是让每个边界能够被独立理解和测试。总结FastAPI 工程化的核心不是堆组件而是明确生命周期和边界配置对象负责从环境加载配置依赖管理请求级资源Service 控制业务与事务Repository 隔离数据访问认证依赖解析当前身份中间件处理跨接口能力Lifespan 管理进程级资源测试通过依赖覆盖隔离外部系统部署层解决多进程、代理、监控和故障恢复。从 Open WebUI、Langflow 和官方模板可以看到FastAPI 的价值不仅是快速写出接口更在于它允许项目从类型声明和依赖注入开始平滑演进到具备数据库、认证、测试、可观测性与复杂生命周期的生产系统。返回上篇FastAPI 实战教程上核心机制、项目配置与完整请求链路参考资料SQL Databaseshttps://fastapi.tiangolo.com/tutorial/sql-databases/Securityhttps://fastapi.tiangolo.com/tutorial/security/Middlewarehttps://fastapi.tiangolo.com/tutorial/middleware/Background Taskshttps://fastapi.tiangolo.com/tutorial/background-tasks/Lifespanhttps://fastapi.tiangolo.com/advanced/events/Testinghttps://fastapi.tiangolo.com/tutorial/testing/Deployment Conceptshttps://fastapi.tiangolo.com/deployment/concepts/Open WebUIhttps://github.com/open-webui/open-webuiLangflowhttps://github.com/langflow-ai/langflowFull Stack FastAPI Templatehttps://github.com/fastapi/full-stack-fastapi-template