如何让FastAPI主应用的异常处理器作用于挂载子应用?
问题原因
FastAPI中通过mount挂载的子应用是独立的ASGI实例,请求进入子应用后,会由子应用自身的异常处理链处理错误,不会自动回退到主应用的异常处理器,所以主应用定义的处理器无法作用于子应用。
解决方案
以下几种方式可以避免重复定义异常处理器,让主应用的逻辑复用在子应用上:
方案1:批量给子应用同步主应用的异常处理器
写一个工具函数,一次性把主应用的所有异常处理器注册到子应用上:
def sync_main_exception_handlers(main_app: FastAPI, *sub_apps: FastAPI): for sub_app in sub_apps: # 合并主应用的异常处理器到子应用,子应用已有相同类型的处理器会被覆盖 sub_app.exception_handlers.update(main_app.exception_handlers) # 挂载子应用后调用该函数 sync_main_exception_handlers(app, mounted_app_1, mounted_app_2)
方案2:自定义基类FastAPI,统一内置异常处理器
如果所有子应用都是你自己创建的FastAPI实例,可以先定义一个带异常处理器的基类,所有应用(包括主应用)都基于这个基类创建:
from pydantic import ValidationError from fastapi import FastAPI, Request from fastapi.responses import JSONResponse class BaseFastAPI(FastAPI): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 在这里统一定义全局异常处理器 @self.exception_handler(ValidationError) async def request_validation_error_exception_handler( request: Request, exc: ValidationError ): return JSONResponse( status_code=422, content={"message": "validation error"}, ) # 主应用和子应用都用这个基类创建 app = BaseFastAPI() mounted_app_1 = BaseFastAPI() mounted_app_2 = BaseFastAPI()
这样所有应用都会自带相同的异常处理逻辑,无需重复编写。
方案3:仅补充子应用缺失的异常处理器
如果子应用已有部分自定义异常处理器,只想补充它没有的主应用处理器,可以用以下方式:
for sub_app in [mounted_app_1, mounted_app_2]: for exc_type, handler in app.exception_handlers.items(): # 只添加子应用未定义的异常类型处理器 if exc_type not in sub_app.exception_handlers: sub_app.exception_handlers[exc_type] = handler
内容的提问来源于stack exchange,提问作者William Pollet
相关产品推荐
相关产品推荐

