文章目录前言一、阶段学习目标二、核心概念什么是中间件2\.1 执行流程2\.2 两大中间件写法2\.3 适用业务场景三、实战 1CORS 跨域配置前后端分离必备3\.1 同源策略原理3\.2 基础配置开发环境3\.3 生产环境安全规范避坑重点四、实战 2自定义通用中间件4\.1 简易装饰器中间件请求耗时 \ 日志4\.2 进阶类中间件全局 Request\-ID 链路追踪4\.3 组合注册统一在 core/middlewares 管理五、实战 3APIRouter 模块化拆分大型项目核心5\.1 单业务路由示例 routers/\[user\.py\]\(user\.py\)5\.2 路由聚合 routers/**init**\.py5\.3 main 统一挂载六、企业级大型标准项目分层架构6\.1 完整目录适配 SQLModel、JWT、中间件、迁移6\.2 各目录职责分层核心意义隔离变更6\.3 \[main\.py\]\(main\.py\) 统一初始化模板七、阶段综合实战完整代码八、新手高频避坑指南九、阶段核心总结前言前面我们已经完成数据库、JWT 鉴权、依赖注入、CRUD 全套业务能力但小型单文件代码无法支撑多人协作、大型后台项目。本阶段解决三大工程核心痛点跨域 CORS前后端分离 Vue/React 前端请求接口 403 拦截生产安全规范配置自定义中间件全局请求日志、耗时统计、统一请求 ID、IP 拦截、全局校验大型分层架构单文件拆分模块化APIRouter、多版本 API、core 核心包规范适配企业多人开发。学习前置已掌握 FastAPI 路由、依赖注入、SQLModel 数据库本阶段完全贴合线上项目标准1 天学完可直接搭建可维护大型后端骨架。一、阶段学习目标理解中间件执行流程请求前置处理、响应后置处理掌握官方CORSMiddleware跨域配置区分开发 / 生产安全写法编写通用异步中间件请求耗时日志、全局请求 ID、客户端 IP 记录学会 Starlette 基类中间件实现复杂拦截逻辑黑白名单限流掌握APIRouter模块化拆分按业务拆分路由、统一版本前缀企业标准分层目录core 核心、api 路由、db 数据库、modules 业务隔离全局统一注册中间件、异常、路由、数据库初始化集中管理多人协作项目规范分层职责、代码解耦、变更隔离。二、核心概念什么是中间件2.1 执行流程所有 HTTP 请求都会先经过中间件再进入路由函数路由处理完成后再次经过中间件返回响应。执行顺序客户端请求 → 中间件前置逻辑 → 路由接口执行 → 中间件后置逻辑 → 返回前端2.2 两大中间件写法装饰器简易app.middleware(http)快速实现轻量逻辑日志、耗时继承BaseHTTPMiddleware类适合复杂、可复用中间件限流、黑白名单。2.3 适用业务场景前后端跨域 CORS 处理全局请求日志、耗时监控生成全局 Request-ID日志链路追踪黑白 IP 拦截、接口限流统一响应头添加、全局权限预检全局异常兜底补充处理。三、实战 1CORS 跨域配置前后端分离必备3.1 同源策略原理前端页面域名、端口、协议任意一项与后端不一致浏览器触发跨域拦截Vue/React 开发环境localhost:5173对接后端8000必出现跨域报错。3.2 基础配置开发环境fromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddleware appFastAPI()# 允许的前端域名列表allow_origins[http://localhost:5173,http://127.0.0.1:5173]app.add_middleware(CORSMiddleware,allow_originsallow_origins,# 允许来源allow_credentialsTrue,# 允许携带Cookie/Token凭证allow_methods[*],# 允许所有请求方法 GET/POST/PUT/DELETEallow_headers[*],# 允许所有请求头Authorizationmax_age86400# 预检请求缓存1天减少OPTIONS请求)3.3 生产环境安全规范避坑重点❌ 禁止生产使用allow_origins[*]通配符无法搭配allow_credentialsTrueToken/Cookie 会失效存在跨站风险。✅ 生产从.env 读取域名配置硬编码改为配置中心# config/settings.py 读取配置classGlobalSettings(BaseSettings):cors_origins:list[str]model_configSettingsConfigDict(env_file.env)# .envCORS_ORIGINShttp://xxx.com,https://admin.xxx.com# main.py 引入配置app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[GET,POST,PUT,DELETE],# 限制危险方法allow_headers[Content-Type,Authorization],# 仅开放必要请求头max_age86400)四、实战 2自定义通用中间件4.1 简易装饰器中间件请求耗时 日志importtimefromfastapiimportRequest# 定义 HTTP 中间件拦截并处理所有进入应用的 HTTP 请求app.middleware(http)asyncdeflog_middleware(request:Request,call_next):# 请求前置逻辑在路由处理函数执行前运行 # 记录请求开始的高精度时间用于后续计算耗时starttime.perf_counter()# 提取并记录请求的基本信息client_iprequest.client.host# 获取客户端 IP 地址methodrequest.method# 获取 HTTP 请求方法如 GET, POSTpathrequest.url.path# 获取请求的 URL 路径# 打印请求进入的日志信息print(f【请求进入】{method}{path}客户端IP:{client_ip})# 调用下一个中间件或路由处理函数获取响应对象responseawaitcall_next(request)# 响应后置逻辑在路由处理函数执行后运行 # 计算请求总耗时毫秒并保留两位小数costround((time.perf_counter()-start)*1000,2)# 在响应头中添加自定义字段方便前端或监控系统获取接口性能数据response.headers[X-Request-Cost-ms]str(cost)# 打印请求处理完成的日志包含状态码和耗时print(f【请求结束】{method}{path}状态码:{response.status_code}耗时:{cost}ms)# 将修改后的响应对象返回给客户端returnresponse4.2 进阶类中间件全局 Request-ID 链路追踪importuuidfromstarlette.middleware.baseimportBaseHTTPMiddlewarefromstarlette.requestsimportRequestfromstarlette.responsesimportResponseclassRequestIdMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request:Request,call_next)-Response:# 从请求头获取或生成新request-idreq_idrequest.headers.get(X-Request-ID,str(uuid.uuid4())[:16])# 传递给后续所有日志、异常responseawaitcall_next(request)# 响应返回request-id用于前后端排查问题response.headers[X-Request-ID]req_idreturnresponse# 注册中间件app.add_middleware(RequestIdMiddleware)4.3 组合注册统一在 core/middlewares 管理企业项目不建议写在 main单独文件统一管理所有中间件集中注册# core/middlewares.pyfromfastapi.middleware.corsimportCORSMiddlewaredefregister_middlewares(app:FastAPI,settings):# 1. CORS跨域app.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)# 2 请求ID中间件app.add_middleware(RequestIdMiddleware)# 3 日志耗时中间件装饰器需封装函数注册app.middleware(http)asyncdeflog_mid(request,call_next):...五、实战 3APIRouter 模块化拆分大型项目核心单文件路由臃肿使用APIRouter按业务拆分用户、订单、管理员、公共接口支持统一前缀、分组标签、模块专属依赖。5.1 单业务路由示例 routers/[user.py](user.py)fromfastapiimportAPIRouter,Dependsfromdependencies.authimportget_active_user# 独立路由统一前缀、标签、模块登录依赖user_routerAPIRouter(prefix/users,tags[用户管理],dependencies[Depends(get_active_user)]# 当前模块全部接口需要登录)user_router.get(/me,summary获取个人信息)defget_info():return{}5.2 路由聚合 routers/init.py统一收集所有业务路由方便 main 导入fromfastapiimportAPIRouterfromrouters.userimportuser_routerfromrouters.orderimportorder_routerfromrouters.authimportauth_router# v1版本总路由api_v1APIRouter(prefix/api/v1)api_v1.include_router(auth_router)api_v1.include_router(user_router)api_v1.include_router(order_router)5.3 main 统一挂载fromroutersimportapi_v1 app.include_router(api_v1)六、企业级大型标准项目分层架构6.1 完整目录适配 SQLModel、JWT、中间件、迁移fastapi_big_project/ ├── .env # 环境变量配置 ├── alembic/ # 数据库迁移 ├── alembic.ini ├── requirements.txt ├── main.py # 程序入口、统一注册组件 ├── core/ # 全局核心无业务 │ ├── __init__.py │ ├── config.py # pydantic-settings全局配置 │ ├── middlewares.py # CORS/日志/RequestID所有中间件 │ ├── exceptions.py # 全局异常捕获、统一返回 │ ├── logger.py # 日志格式化配置 │ └── dependencies.py # 全局公共依赖分页、鉴权 ├── db/ # 数据库全局 │ ├── __init__.py │ └── session.py # engine、get_db会话依赖 ├── models/ # SQLModel数据库实体 │ ├── user.py │ └── order.py ├── schemas/ # DTO分层模型 │ ├── user_schema.py │ └── order_schema.py ├── crud/ # 通用CRUD业务操作 │ ├── base.py │ ├── user_crud.py │ └── order_crud.py ├── routers/ # 模块化路由 │ ├── __init__.py # v1总路由聚合 │ ├── auth_router.py │ ├── user_router.py │ └── order_router.py └── utils/ # 工具函数加密、JWT、日期 ├── __init__.py └── security.py6.2 各目录职责分层核心意义隔离变更目录职责修改影响范围core全局配置、中间件、异常、通用依赖全项目生效修改需全量测试db数据库引擎、会话仅数据库连接逻辑models数据表实体仅表结构配合 Alembic 迁移schemas入参 / 返回 DTO接口入参、返回字段调整crud数据库业务逻辑增删改查 SQL 逻辑routersAPI 路由、接口参数仅 URL、请求方式、入参utils通用工具函数加密、时间、校验工具6.3 [main.py](main.py) 统一初始化模板所有中间件、异常、路由、数据库集中注册入口极简干净fromfastapiimportFastAPIfromcore.configimportsettingsfromcore.middlewaresimportregister_middlewaresfromcore.exceptionsimportregister_exceptionfromdb.sessionimportenginefromsqlmodelimportSQLModelfromroutersimportapi_v1# 开发临时建表definit_tables():ifsettings.app_envdev:SQLModel.metadata.create_all(bindengine)# 创建应用实例defcreate_app()-FastAPI:appFastAPI(title大型后台管理系统,version1.0)# 1 注册中间件CORS、日志register_middlewares(app,settings)# 2 注册全局异常处理器register_exception(app)# 3 挂载路由app.include_router(api_v1)# 4 启动事件初始化数据库app.on_event(startup)defstartup():init_tables()returnapp appcreate()if__name____main__:importuvicorn uvicorn.run(main:app,reloadTrue,host0.0.0.0,port8000)七、阶段综合实战完整代码整合 CORS、请求日志中间件、模块化路由、分层目录入口# core/middlewares.pyimporttimeimportuuidfromfastapiimportFastAPI,Requestfromfastapi.middleware.corsimportCORSMiddlewarefromstarlette.middleware.baseimportBaseHTTPMiddlewareclassRequestIdMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):ridrequest.headers.get(X-Request-ID,str(uuid.uuid4())[:12])respawaitcall_next(request)resp.headers[X-Request-ID]ridreturnrespdefregister_middlewares(app:FastAPI,settings):# CORSapp.add_middleware(CORSMiddleware,allow_originssettings.cors_origins,allow_credentialsTrue,allow_methods[*],allow_headers[*],)# RequestIDapp.add_middleware(RequestIdMiddleware)# 耗时日志中间件app.middleware(http)asyncdeflog_mid(req:Request,call_next):stime.perf_counter()respawaitcall_next(req)msround((time.perf_counter()-s)*1000,2)resp.headers[X-Cost]str(ms)print(f{req.method}{req.url.path}{resp.status_code}{ms}ms)returnresp八、新手高频避坑指南❌ 生产CORS使用allow_origins[*]携带 Token/Cookie 失效❌ 所有路由写在 main项目膨胀后无法维护❌ 中间件注册顺序混乱CORS 放在日志中间件后面导致 OPTIONS 无跨域头❌ 中间件忘记await call_next(request)请求卡死无响应❌ 业务、配置、数据库代码全部混在同一文件多人协作冲突严重❌ 中间件内读取请求 body 导致流消耗后续接口拿不到参数✅ CORS 域名从环境变量读取区分开发生产✅ 按业务拆分 APIRouter统一版本前缀/api/v1✅ 中间件统一封装在 core不在 main 堆代码✅ 分层严格遵守职责数据库逻辑不写路由。九、阶段核心总结中间件分装饰器简易版、类复杂版前置拦截 后置统一处理CORS前后端分离必备开发通配生产限定域名禁止[*]搭配凭证模块化路由APIRouter拆分业务统一前缀、模块专属登录依赖大型分层core 核心、db 数据库、models 实体、schemas 入参、crud 业务、routers 接口完全解耦工程规范main 仅做统一注册所有全局组件集中初始化业务代码隔离。
)
