从CORS到自定义，让你的API更健壮

一、中间件是啥咱用“餐厅”打个比方想象一下你的FastAPI应用是个高级餐厅。 顾客客户端请求来到门口。-迎宾CORS中间件先看你是不是从允许的街区域名来的不是就直接劝退返回CORS错误。-领位员日志中间件记下顾客几点来的、几个人。-安检认证中间件检查你有没有预约码Token。 之后顾客才能见到真正的厨师你的路径操作函数点菜吃饭处理业务逻辑。 吃完离开时还得经过传菜员响应处理中间件可能给打包盒贴个标签添加响应头。这一系列站在“核心业务”前后的服务人员就是中间件。它的核心价值在于全局处理、业务无侵入。官方文档讲得比较散咱们今天就把这块硬骨头啃透直接上代码。 二、核心怎么用从最常用的CORS开始好咱们先来解决开头的“跨域”问题。这是99%的Web应用都会遇到的。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware app FastAPI() # 1. 配置允许的源、方法、请求头 origins [ http://localhost:3000, # 你的前端开发地址 https://your-production-site.com, ] # 2. 添加中间件 app.add_middleware( CORSMiddleware, allow_originsorigins, # 允许的源列表也可以用 [*] 放行所有不安全 allow_credentialsTrue, # 允许携带Cookie allow_methods[*], # 允许所有方法 (GET, POST, 等) allow_headers[*], # 允许所有请求头 ) app.get(/) async def main(): return {message: Hello World}这里千万别学我当初偷懒图省事直接上 allow_origins[*]这在生产环境是安全大忌相当于餐厅大门敞开谁都能进。线上务必明确指定前端域名 三、动手写一个自己的中间件接下来重点来了自定义中间件。比如我们要给每个请求记日志并计算处理耗时。import time from fastapi import FastAPI, Request app FastAPI() app.middleware(http) # 这是关键装饰器 async def log_request_time(request: Request, call_next): # 1. 请求进来时 # 这里我们使用 time.perf_counter() 而不是 time.time()因为它对于这些用例可能更精确 start_time time.perf_counter() path request.url.path method request.method print(f 收到请求: {method} {path}) # 2. 把请求交给下一个处理环节可能是其他中间件或者是最终的路由 response await call_next(request) # 3. 响应返回前 process_time time.perf_counter() - start_time response.headers[X-Process-Time] str(process_time) # 可以往响应头加东西 print(f✅ 请求完成: {method} {path}, 耗时: {process_time:.4f}秒) return response app.get(/test) async def test(): return {message: ok}访问/test看看控制台是不是日志和响应头都有了这就是一个最基础的HTTP中间件。划重点中间件函数接收一个call_next它就像接力棒你必须调用它await call_next(request)请求才能继续往后走。你在它前面和后面写的代码就分别对应了“请求处理”和“响应处理”两个阶段。 四、灵魂拷问多个中间件谁先谁后你是不是以为加完就完事了多个中间件的执行顺序是超级易错点想象一下你既加了CORS中间件又加了上面的日志中间件还加了一个认证中间件。它们怎么排队结论按照添加的相反顺序执行“请求阶段”再按照添加的正序执行“响应阶段”。像洋葱一样一层层进去再一层层出来。app FastAPI() # 假设我们按顺序添加三个中间件 app.add_middleware(MiddlewareC) # 第三个添加 app.add_middleware(MiddlewareB) # 第二个添加 app.middleware(http)(middleware_a) # 第一个添加装饰器写法 # 实际执行顺序请求阶段 # 1. middleware_a 的请求处理代码 # 2. MiddlewareB 的请求处理代码 # 3. MiddlewareC 的请求处理代码 # --- 到达路由函数 --- # 4. MiddlewareC 的响应处理代码 # 5. MiddlewareB 的响应处理代码 # 6. middleware_a 的响应处理代码 # --- 响应返回给客户端 ---官方文档虽然说了是“装饰器顺序”但用app.add_middleware()添加时更容易迷糊。记不住就背下这个口诀“后来居上请求原路返回响应”。设计时要把依赖关系想清楚比如认证应该放在靠“里”层后添加日志可以放在最“外”层先添加。 五、进阶更底层的ASGI中间件再说个容易翻车的点。上面我们用app.middleware(http)叫HTTP中间件是FastAPI封装好的。还有一种更底层、更强大的叫ASGI中间件。它和HTTP中间件啥区别好比一个是高级餐厅的固定流程HTTP另一个是后厨的原子操作ASGI能处理WebSocket等更多协议。怎么用通常你需要一个第三方库比如starlette-context来在请求中传递全局数据或者自己封装FastAPI在fastapi.middleware中提供了几个中间件仅仅是为了方便开发者但大多数可用的中间件直接来自Starlettefrom fastapi import FastAPI from starlette.middleware.base import BaseHTTPMiddleware class CustomHeaderMiddleware(BaseHTTPMiddleware): async def dispatch(self, request, call_next): response await call_next(request) response.headers[X-Custom-Header] MyValue return response app FastAPI() app.add_middleware(CustomHeaderMiddleware) # 这里添加的就是ASGI中间件注意BaseHTTPMiddleware使用简单但官方提示可能有轻微性能损耗因为每个请求都会创建新的类实例。对于超高并发场景用之前的函数式app.middleware(http)或直接写纯ASGI中间件是更优解。 六、我的踩坑经验与最佳实践最后啰嗦几句都是实战中总结的血泪教训1️⃣响应式中间件要小心如果你想在中间件里读取响应的body比如做全局响应体格式化需要先把它读出来body await response.body()但这会让响应体在内存中多存一份。对于大文件流响应慎用2️⃣异常处理要覆盖中间件里的call_next可能会抛出异常记得用try...except包住并返回一个合理的错误响应而不是让服务崩掉。3️⃣区分中间件和依赖注入像认证这种每个路由可能要求不同有的需要登录有的不需要用依赖注入Depends更灵活。中间件更适合做全局的、无差异的处理比如CORS、全局日志、GZip压缩。工具的选择好比选螺丝刀不是最贵的就好而是“合适”最好。对于大部分业务场景HTTP中间件依赖注入的组合已经能解决99%的问题了。