拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线：400-886-1026 service@lmnt.cn
拓冰网站定制 拓冰网站定制TOBE WEB STUDIO
免费定制
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

首页/资讯中心/资讯详情

FastAPI 实战教程（下）：数据库、JWT、测试部署与热门项目架构拆解

发布时间：2026/10/5 6:59:47 拓冰网站定制 建站干货
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 → Database

Router 负责 HTTP 协议，Service 负责业务规则，Repository 负责数据访问。不要让路由函数同时完成参数校验、SQL、事务、邮件发送和权限判断，否则测试与复用都会迅速变难。

二、配置管理：不要在代码里散落环境变量

安装设置组件：

pip install pydantic-settings

app/core/config.py：

fromfunctoolsimportlru_cachefrompydanticimportSecretStrfrompydantic_settingsimportBaseSettings,SettingsConfigDictclassSettings(BaseSettings):app_name:str="FastAPI Service"environment:str="development"database_url:str="sqlite:///./app.db"jwt_secret:SecretStr access_token_minutes:int=30cors_origins:list[str]=["http://localhost:5173"]model_config=SettingsConfigDict(env_file=".env",env_file_encoding="utf-8",case_sensitive=False,extra="ignore",)@lru_cachedefget_settings()->Settings:returnSettings()

.env：

APP_NAME=FastAPI Demo ENVIRONMENT=development DATABASE_URL=postgresql+psycopg://app:password@localhost:5432/app JWT_SECRET=请替换为随机长字符串 ACCESS_TOKEN_MINUTES=30 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.py：

fromcollections.abcimportGeneratorfromsqlmodelimportSession,create_enginefromapp.core.configimportget_settings settings=get_settings()engine=create_engine(settings.database_url,pool_pre_ping=True,)defget_session()->Generator[Session,None,None]:withSession(engine)assession:yieldsession

这里的yield与请求生命周期绑定：进入路径函数前创建 Session，请求结束后关闭。依赖函数本身不应该自动 commit，因为一个业务操作可能包含多次写入，事务边界更适合放在 Service 层。

模型与 Schema：

fromdatetimeimportdatetimefrompydanticimportBaseModel,ConfigDictfromsqlmodelimportField,SQLModelclassTask(SQLModel,table=True):id:int|None=Field(default=None,primary_key=True)title:str=Field(index=True,max_length=200)completed:bool=Falsecreated_at:datetime=Field(default_factory=datetime.utcnow)classTaskCreate(BaseModel):title:strclassTaskPublic(BaseModel):model_config=ConfigDict(from_attributes=True)id:inttitle:strcompleted:boolcreated_at:datetime

Repository：

fromsqlmodelimportSession,selectfromapp.db.modelsimportTaskclassTaskRepository:def__init__(self,session:Session):self.session=sessiondeflist(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)returntask

Service 决定事务：

classTaskService:def__init__(self,session:Session):self.session=session self.repo=TaskRepository(session)defcreate(self,title:str)->Task:ifnottitle.strip():raiseValueError("title cannot be empty")try:task=self.repo.add(Task(title=title.strip()))self.session.commit()returntaskexceptException:self.session.rollback()raise

路由只负责协议转换：

fromtypingimportAnnotatedfromfastapiimportAPIRouter,Depends,statusfromsqlmodelimportSession router=APIRouter(prefix="/tasks",tags=["tasks"])SessionDep=Annotated[Session,Depends(get_session)]@router.post("",response_model=TaskPublic,status_code=status.HTTP_201_CREATED)defcreate_task(payload:TaskCreate,session:SessionDep):returnTaskService(session).create(payload.title)

同步 SQLAlchemy/SQLModel 应配合同步def路由，FastAPI 会将其放入线程池。如果选择 SQLAlchemy AsyncSession，则应从驱动、Session 到 Repository 全链路异步，不要混用。

四、认证：OAuth2PasswordBearer 只是取 Token

OAuth2PasswordBearer不会自动验证用户，它主要完成两件事：告诉 OpenAPI 使用 Bearer Token，并从 Authorization Header 提取 Token。

fromtypingimportAnnotatedfromfastapiimportDependsfromfastapi.securityimportOAuth2PasswordBearer oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")TokenDep=Annotated[str,Depends(oauth2_scheme)]

安装密码和 JWT 工具：

pip install"pwdlib[argon2]"pyjwt

app/core/security.py：

fromdatetimeimportdatetime,timedelta,timezoneimportjwtfrompwdlibimportPasswordHashfromapp.core.configimportget_settings password_hash=PasswordHash.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:settings=get_settings()now=datetime.now(timezone.utc)payload={"sub":subject,"iat":now,"exp":now+timedelta(minutes=settings.access_token_minutes),}returnjwt.encode(payload,settings.jwt_secret.get_secret_value(),algorithm="HS256",)

解析当前用户：

fromfastapiimportHTTPException,statusfromjwtimportInvalidTokenErrorasyncdefget_current_user(token:TokenDep,session:SessionDep):credentials_error=HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Could not validate credentials",headers={"WWW-Authenticate":"Bearer"},)try:payload=jwt.decode(token,get_settings().jwt_secret.get_secret_value(),algorithms=["HS256"],)user_id=int(payload["sub"])except(InvalidTokenError,KeyError,ValueError):raisecredentials_error user=session.get(User,user_id)ifuserisNoneornotuser.is_active:raisecredentials_errorreturnuser

权限可以继续建立在当前用户依赖上：

defrequire_admin(user:CurrentUser):ifnotuser.is_admin:raiseHTTPException(status_code=403,detail="Admin required")returnuser

认证是 401，已登录但权限不足是 403；两者不要混用。

五、中间件与 CORS

fromfastapi.middleware.corsimportCORSMiddleware settings=get_settings()app.add_middleware(CORSMiddleware,allow_origins=settings.cors_origins,allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)

如果allow_credentials=True，生产环境不要简单使用allow_origins=["*"]，应明确列出可信前端来源。

自定义请求耗时中间件：

importtimeimportuuidfromfastapiimportRequest@app.middleware("http")asyncdefrequest_context(request:Request,call_next):request_id=request.headers.get("X-Request-ID",str(uuid.uuid4()))started=time.perf_counter()response=awaitcall_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",encoding="utf-8")asfile:file.write(f"created task{task_id}\n")@router.post("",response_model=TaskPublic)defcreate_task(payload:TaskCreate,session:SessionDep,background_tasks:BackgroundTasks,):task=TaskService(session).create(payload.title)background_tasks.add_task(write_audit_log,task.id)returntask

BackgroundTasks 会在响应发送后、当前应用进程中执行。它适合短小、失败可容忍的工作。视频转码、大批量邮件、模型推理等长任务应使用 Celery、Dramatiq、RQ 或独立消息队列，因为进程重启会丢失内存中的后台任务。

七、Lifespan：初始化和释放共享资源

fromcontextlibimportasynccontextmanagerfromfastapiimportFastAPI@asynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.http_client=AsyncClient(timeout=10)app.state.model=awaitload_model()yieldawaitapp.state.http_client.aclose()awaitapp.state.model.close()app=FastAPI(lifespan=lifespan)

yield前只执行一次启动逻辑，yield后执行关闭逻辑。数据库连接池、HTTP Client、模型和缓存客户端适合在这里管理。不要在 import 模块时执行昂贵初始化，否则测试收集、CLI 工具和多 Worker 启动都会受到影响。

八、测试：覆盖依赖边界，而不是启动真实服务器

FastAPI 的 TestClient 基于 HTTPX：

fromfastapi.testclientimportTestClientfromapp.mainimportapp client=TestClient(app)deftest_create_task():response=client.post("/api/v1/tasks",json={"title":"write tests"})assertresponse.status_code==201assertresponse.json()["title"]=="write tests"

认证依赖可以替换：

deffake_current_user():returnUser(id=1,email="tester@example.com",is_active=True)app.dependency_overrides[get_current_user]=fake_current_userdeftest_private_endpoint():response=client.get("/api/v1/profile")assertresponse.status_code==200defteardown_module():app.dependency_overrides.clear()

依赖覆盖比在测试中签发真实 JWT 更适合路由单元测试；认证编解码本身再使用独立测试覆盖。

异步测试可以使用 HTTPX AsyncClient 和 ASGITransport：

importpytestfromhttpximportASGITransport,AsyncClient@pytest.mark.anyioasyncdeftest_root():transport=ASGITransport(app=app)asyncwithAsyncClient(transport=transport,base_url="http://test",)asclient:response=awaitclient.get("/")assertresponse.status_code==200

九、生产部署需要考虑什么

本地开发：

fastapi dev app/main.py

单进程生产启动：

fastapi run app/main.py--host 0.0.0.0--port 8000

使用 Uvicorn：

uvicorn app.main:app--host0.0.0.0--port8000--workers4

Worker 数不是越多越好。每个 Worker 都是独立进程，会分别创建连接池、缓存和模型对象，应结合 CPU、内存与压测结果决定。

一个基础 Dockerfile：

FROM 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.py

fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromapp.api.routesimportauth,tasks,usersfromapp.core.configimportget_settingsfromapp.lifecycleimportlifespandefcreate_app()->FastAPI:settings=get_settings()app=FastAPI(title=settings.app_name,lifespan=lifespan,)app.add_middleware(CORSMiddleware,allow_origins=settings.cors_origins,allow_credentials=True,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 app=create_app()

Application Factory 让测试可以在不同配置下创建应用，也把路由、中间件和生命周期的装配位置固定下来。

十一、从热门项目反推生产级架构

前面的工程结构不是凭空设计出来的。下面选择三个采用 FastAPI 的代表性仓库，重点看它们的应用创建、生命周期、路由组织和数据边界，而不是复述 README。

1. Open WebUI：让 FastAPI 成为 AI 平台的统一入口

Open WebUI 的后端不仅提供普通 REST API，还要协调模型连接、认证、知识库、文件、WebSocket 和前端静态资源。它在创建应用时显式传入lifespan，并根据环境决定是否暴露文档：

app=FastAPI(docs_url="/docs"ifENV=="dev"elseNone,openapi_url="/openapi.json"ifENV=="dev"elseNone,lifespan=lifespan,)

这种写法包含两个重要判断：生产环境不必默认暴露交互式文档；连接池、模型客户端和缓存等共享资源应该交给生命周期统一创建和释放，而不是散落在模块导入阶段。

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. Langflow：Application Factory 与复杂 Lifespan

Langflow 需要加载组件、数据库、缓存、可观测性和执行引擎。它采用创建函数组织应用，使测试、命令行和部署入口可以用不同配置得到 FastAPI 实例：

defcreate_app()->FastAPI:app=FastAPI(lifespan=lifespan)app.include_router(api_router,prefix="/api/v1")register_exception_handlers(app)configure_cors(app)returnapp

复杂启动逻辑被放进lifespan，并使用yield分隔初始化与清理：

@asynccontextmanagerasyncdeflifespan(app:FastAPI):app.state.services=awaitbuild_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:EmailStr=Field(unique=True,index=True)is_active:bool=TrueclassUserCreate(UserBase):password:str=Field(min_length=8,max_length=128)classUserPublic(UserBase):id:UUID

第二，Session 和当前用户通过依赖传入，路由不自行创建数据库连接：

SessionDep=Annotated[Session,Depends(get_db)]CurrentUser=Annotated[User,Depends(get_current_user)]@router.get("/me",response_model=UserPublic)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 Databases：https://fastapi.tiangolo.com/tutorial/sql-databases/
  • Security：https://fastapi.tiangolo.com/tutorial/security/
  • Middleware：https://fastapi.tiangolo.com/tutorial/middleware/
  • Background Tasks：https://fastapi.tiangolo.com/tutorial/background-tasks/
  • Lifespan：https://fastapi.tiangolo.com/advanced/events/
  • Testing：https://fastapi.tiangolo.com/tutorial/testing/
  • Deployment Concepts：https://fastapi.tiangolo.com/deployment/concepts/
  • Open WebUI：https://github.com/open-webui/open-webui
  • Langflow：https://github.com/langflow-ai/langflow
  • Full Stack FastAPI Template：https://github.com/fastapi/full-stack-fastapi-template
网站建设 企业官网 运营推广 建站干货
返回列表