Starlette 异常处理完全指南自定义异常处理器、HTTPException 与 WebSocketException 实战【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette导读Starlette 允许你安装自定义的异常处理器exception handlers用于决定当错误或可处理异常发生时如何返回响应。本文以 docs/exceptions.md 为骨架结合仓库源码starlette/exceptions.py、starlette/middleware/exceptions.py、starlette/middleware/errors.py 等与测试用例tests/test_exceptions.py系统讲解按状态码/异常类注册处理器、区分错误与可处理异常、HTTP 与 WebSocket 场景下的异常处理等核心知识。读完本文你将掌握如何为 404/500 定制响应、如何将HTTPException输出为 JSON、如何处理 WebSocket 异常并深入理解 Starlette 内部异常处理中间件的工作机制。一、自定义异常处理器入门Starlette 允许你安装自定义的异常处理器来决定当错误或已处理异常发生时如何返回响应。异常处理器是一个以(conn, exc) - response为签名的可调用对象第一个参数是当前连接对象HTTP 请求为RequestWebSocket 连接为WebSocket第二个参数是被捕获的异常实例返回值是一个Response对象WebSocket 处理器可以返回None见下文。最简单的做法是在构造Starlette应用时通过exception_handlers参数传入一个映射键可以是整数状态码也可以是异常类from starlette.applications import Starlette from starlette.exceptions import HTTPException from starlette.requests import Request from starlette.responses import HTMLResponse HTML_404_PAGE ... HTML_500_PAGE ... async def not_found(request: Request, exc: HTTPException): return HTMLResponse(contentHTML_404_PAGE, status_codeexc.status_code) async def server_error(request: Request, exc: HTTPException): return HTMLResponse(contentHTML_500_PAGE, status_codeexc.status_code) exception_handlers { 404: not_found, 500: server_error } app Starlette(routesroutes, exception_handlersexception_handlers)从 starlette/applications.py 的Starlette.__init__签名可以看到exception_handlers的类型为Mapping[Any, ExceptionHandler]即键既可以是int状态码也可以是type[Exception]异常类。debug 模式下的差异如果开启了debug并且发生错误Starlette 将不再使用安装的 500 处理器而是直接返回一个 traceback 响应调试用app Starlette(debugTrue, routesroutes, exception_handlersexception_handlers)该行为的实现在 starlette/middleware/errors.py 的ServerErrorMiddleware.__call__中当self.debug为真时调用self.debug_response(request, exc)。debug_response会根据请求的Accept头决定返回类型——客户端接受text/html时返回带样式、可折叠帧、带源码高亮的 HTML 调试页对应文件中的STYLES/JS/TEMPLATE等模板常量否则返回纯文本 tracebackgenerate_plain_text。若debugFalse且未安装 500 处理器则返回默认的纯文本Internal Server Errorerror_response方法。二、按异常类注册处理器除了为特定的状态码注册处理器你还可以为异常类注册处理器。特别是你很可能想覆盖内置的HTTPException类的处理方式。例如改用 JSON 风格的响应async def http_exception(request: Request, exc: HTTPException): return JSONResponse({detail: exc.detail}, status_codeexc.status_code) exception_handlers { HTTPException: http_exception }处理器查找机制基于 MRO 的类型匹配为什么按异常类注册能覆盖所有HTTPException及其子类答案在 starlette/_exception_handler.py 的_lookup_exception_handlerdef _lookup_exception_handler(exc_handlers: ExceptionHandlers, exc: Exception) - ExceptionHandler | None: for cls in type(exc).__mro__: if cls in exc_handlers: return exc_handlers[cls] return None它遍历抛出异常实例类型的__mro__方法解析顺序即从子类一路向上到Exception第一个命中的类对应的处理器即被选中。这意味着为HTTPException注册处理器会同时接管所有HTTPException子类如果为某个自定义子类如BadBodyException(HTTPException)注册了更具体的处理器则优先命中该子类。这一行为在 tests/test_exceptions.py 中有对应测试自定义的BadBodyException(HTTPException)抛出 422 后由专门的handler_that_reads_body处理并且在处理器中仍然可以读取请求体request.body()因为处理器内复用同一个请求对象。处理器的同步/异步调度wrap_app_handling_exceptionsstarlette/_exception_handler.py通过is_async_callable判断处理器是异步函数还是同步函数异步处理器直接await同步处理器则通过run_in_threadpool放入线程池执行避免阻塞事件循环。因此你可以放心编写同步的异常处理器def sync_handler(request: Request, exc: Exception) - JSONResponse: return JSONResponse({detail: str(exc)}, status_code500)这一设计也有专门测试test_handlers_annotationstests/test_exceptions.py验证同步与异步处理器均能被类型检查器接受。三、HTTPException 的 headers 参数传递HTTPException还配有headers参数它允许把头部信息传播到响应类上async def http_exception(request: Request, exc: HTTPException): return JSONResponse( {detail: exc.detail}, status_codeexc.status_code, headersexc.headers )从 starlette/exceptions.py 的实现看HTTPException.__init__的三个参数为status_code: int——HTTP 状态码detail: str | None None——错误详情若为None则会自动用http.HTTPStatus(status_code).phrase填充即标准状态码短语如 404 对应 Not Foundheaders: Mapping[str, str] | None None——附加响应头。__str__返回f{self.status_code}: {self.detail}__repr__返回HTTPException(status_code..., detail...)这些在 tests/test_exceptions.py 中都有断言。测试 test_with_headers 验证了在端点中raise HTTPException(status_code200, headers{x-potato: always})后响应头x-potato正确透传。另外注意 starlette/middleware/exceptions.py 中默认的http_exception处理器有一个特例当状态码为 204 或 304 时返回不带正文的Response(status_codeexc.status_code, headersexc.headers)其余状态码才返回PlainTextResponse(exc.detail, ...)。对应的测试 test_no_content 断言 204 响应不含content-length头test_not_modified 断言 304 响应正文为空。四、覆盖 WebSocketException 的处理你可能还想覆盖WebSocketException的处理方式async def websocket_exception(websocket: WebSocket, exc: WebSocketException): await websocket.close(code1008) exception_handlers { WebSocketException: websocket_exception }注意 WebSocket 场景下处理器第一个参数是WebSocket对象返回值可以是None此时无需返回 HTTP 响应而是通过websocket.close()关闭连接。默认情况下ExceptionMiddleware内置的websocket_exceptionstarlette/middleware/exceptions.py就是执行await websocket.close(codeexc.code, reasonexc.reason)。WebSocketException 的构造参数starlette/exceptions.py 中WebSocketException的签名为code: int——WebSocket 关闭码reason: str | None None——关闭原因默认空字符串。你可以使用 RFC 6455 第 7.4.1 节 中定义的任何合法关闭码如 1000 正常关闭、1008 策略违规等。__str__与__repr__的实现和对应测试见 tests/test_exceptions.py。五、错误与可处理异常的区分核心概念理解 Starlette 的异常体系关键在于区分可处理异常handled exceptions与错误errors。可处理异常并不代表错误场景。它们会被转换成合适的 HTTP 响应然后穿过标准的中间件栈发送。默认情况下HTTPException类被用来管理所有可处理异常。错误是应用中出现的任何其他异常。这些异常应该作为异常冒泡穿过整个中间件栈。任何错误日志中间件都应确保把异常一路重新抛出re-raise到服务器层。Exception 与 500 两种键的等价性在实际使用中用于处理错误的键是exception_handlers[500]或exception_handlers[Exception]。500和Exception两个键都可以用async def handle_error(request: Request, exc: HTTPException): # Perform some logic return JSONResponse({detail: exc.detail}, status_codeexc.status_code) exception_handlers { Exception: handle_error # or 500: handle_error }这一等价关系的实现在 starlette/applications.py 的build_middleware_stack中for key, value in self.exception_handlers.items(): if key in (500, Exception): error_handler value else: exception_handlers[key] value middleware [Middleware(ServerErrorMiddleware, handlererror_handler, debugdebug)]也就是说键500或Exception的处理器会被提取出来作为ServerErrorMiddleware的handler参数最外层中间件其余键状态码或异常类则传给最内层的ExceptionMiddleware。这从源码层面印证了两键等价且ServerErrorMiddleware只关心 http 类型的 scopestarlette/middleware/errors.py 中非 http scope 直接透传。BackgroundTask 抛异常的特殊语义这里有一个重要的注意点当 BackgroundTask 抛出异常时它会被handle_error函数处理但此时响应早已发送完毕。换句话说handle_error创建的响应会被丢弃。如果错误发生在响应发送之前则会使用返回的响应对象——即上面例子中的JSONResponse。根源在于 starlette/background.py 中BackgroundTask.__call__的实现任务在响应已写出后才被异步执行同步函数还会被丢进线程池。因此后台任务的失败无法通过返回新响应来兜底只能靠日志、监控等手段记录。已发送响应后再抛异常会导致 RuntimeErrorwrap_app_handling_exceptions中用一个response_started标志跟踪http.response.start消息是否已发出。如果异常在响应已经开始后被捕获且存在匹配的处理器则直接抛出RuntimeError(Caught handled exception, but response already started.)starlette/_exception_handler.py。对应的测试 test_handled_exc_after_response 构造了一个先发送 200 响应再抛出 406HTTPException的端点验证默认情况下TestClient会抛出该RuntimeError若使用raise_server_exceptionsFalse客户端仍能收到那个已发送的 200 响应。六、中间件栈的编排方式为了正确处理上述行为一个Starlette应用的中间件栈按如下顺序配置ServerErrorMiddleware——当服务器错误发生时返回 500 响应已安装的中间件用户通过middleware参数或add_middleware添加的中间件ExceptionMiddleware——处理可处理异常并返回响应Router——路由分发Endpoints——最终的业务端点。完整的装配逻辑见 starlette/applications.pymiddleware [Middleware(ServerErrorMiddleware, handlererror_handler, debugdebug)] if self.max_body_size is not None: middleware.append(Middleware(RequestBodyLimitMiddleware, max_body_sizeself.max_body_size)) middleware self.user_middleware middleware.append(Middleware(ExceptionMiddleware, handlersexception_handlers, debugdebug)) app self.router for cls, args, kwargs in reversed(middleware): app cls(app, *args, **kwargs)这样一层层包裹后ServerErrorMiddleware成为最外层任何未被捕获的错误最终都能转换为 500ExceptionMiddleware成为最内层紧贴路由与端点。ServerErrorMiddleware捕获异常后始终会重新抛出starlette/middleware/errors.py以便服务器记录错误、测试客户端选择性地在测试用例中抛出该错误。ExceptionMiddleware 的两个处理器集合starlette/middleware/exceptions.py 中ExceptionMiddleware内部维护两个字典_status_handlers: dict[int, ExceptionHandler]——按状态码索引_exception_handlers: dict[type, ExceptionHandler]——按异常类索引默认内置HTTPException - self.http_exception与WebSocketException - self.websocket_exception。用户传入的处理器通过add_exception_handler分发整数键进_status_handlers异常类键进_exception_handlers并对非Exception子类做断言。调用时这两个字典被放入scope[starlette.exception_handlers]供wrap_app_handling_exceptions读取使用starlette/middleware/exceptions.py。处理顺序上starlette/_exception_handler.py若异常是HTTPException先按exc.status_code在状态码处理器中精确查找若未命中再走_lookup_exception_handler按 MRO 匹配异常类处理器若仍无处理器则原样重新抛出。注意测试 test_http_exception_does_not_use_threadpool 专门验证了默认的http_exception处理器是异步的不会走线程池。七、HTTPException 详解HTTPException类为任何可处理异常提供了一个基类。ExceptionMiddleware的默认实现会对任何HTTPException返回纯文本 HTTP 响应。构造签名HTTPException(status_code, detailNone, headersNone)你只应在路由或端点内部抛出HTTPException。中间件类则应直接返回合适的响应而不是抛异常。这一点与异常处理器的执行位置密切相关HTTPException由最内层的ExceptionMiddleware捕获而中间件位于其外层中间件中抛出的异常不会进入ExceptionMiddleware的处理器查找逻辑。在 WebSocket 端点上使用 HTTPException你可以在 WebSocket 端点上使用HTTPException。如果它在websocket.accept()之前抛出连接将不会被升级为 WebSocket 连接而是返回合适的 HTTP 响应from starlette.applications import Starlette from starlette.exceptions import HTTPException from starlette.routing import WebSocketRoute from starlette.websockets import WebSocket async def websocket_endpoint(websocket: WebSocket): raise HTTPException(status_code400, detailBad request) app Starlette(routes[WebSocketRoute(/ws, websocket_endpoint)])从源码看ExceptionMiddleware.__call__同时处理http与websocket两种 scopestarlette/middleware/exceptions.py并分别构造Request或WebSocket连接对象传给wrap_app_handling_exceptions。另外若要在 accept 之后以 HTTP 响应拒绝 WebSocket 请求可使用 starlette/websockets.py 中WebSocket.send_denial_response依赖服务器的 Websocket Denial Response 扩展支持。八、WebSocketException 详解你可以使用WebSocketException类在 WebSocket 端点内部抛出错误WebSocketException(code, reasonNone)你可以设置 RFC 6455 第 7.4.1 节 中定义的任何合法关闭码。常见的关闭码包括1000正常关闭、1001端点离开、1002协议错误、1003收到无法处理的数据、1008策略违规、1009消息过大、1011内部错误等。当WebSocketException抛出后默认处理器会调用websocket.close(code..., reason...)向客户端发送关闭帧。你也可以像第一节那样注册自定义处理器决定发送哪个关闭码和原因。九、实战组合一个完整的异常处理配置综合以上所有知识这里给出一个集成了状态码、异常类与 WebSocket 异常处理器的完整示例可直接复制运行配合uvicorn等 ASGI 服务器from starlette.applications import Starlette from starlette.exceptions import HTTPException, WebSocketException from starlette.requests import Request from starlette.responses import JSONResponse from starlette.routing import Route, WebSocketRoute from starlette.websockets import WebSocket async def homepage(request: Request): raise HTTPException(status_code404, detailPage not found) async def boom(request: Request): raise RuntimeError(Something went wrong) # 错误而非可处理异常 async def ws_endpoint(websocket: WebSocket): await websocket.accept() raise WebSocketException(code1008, reasonPolicy Violation) async def http_exception(request: Request, exc: HTTPException): # 覆盖内置处理HTTPException 一律输出 JSON return JSONResponse( {detail: exc.detail}, status_codeexc.status_code, headersexc.headers, ) async def websocket_exception(websocket: WebSocket, exc: WebSocketException): await websocket.close(codeexc.code, reasonexc.reason) async def server_error(request: Request, exc: Exception): # 500/Exception 键二选一这里兜底所有未处理错误 return JSONResponse({detail: Internal Server Error}, status_code500) app Starlette( routes[ Route(/, homepage), Route(/boom, boom), WebSocketRoute(/ws, ws_endpoint), ], exception_handlers{ HTTPException: http_exception, # 按异常类注册 WebSocketException: websocket_exception, # WebSocket 专用 Exception: server_error, # 等价于 500: server_error }, # debugTrue 时/boom 将返回 traceback 页面而非 JSON 500 )用 TestClient 验证行为仓库的 tests/test_exceptions.py 展示了如何用TestClient验证异常处理行为例如from starlette.testclient import TestClient # 404 由 http_exception 处理返回 JSON response client.get(/) assert response.status_code 404 assert response.json() {detail: Page not found} # 500 由 server_error 兜底 with TestClient(app, raise_server_exceptionsFalse) as client: response client.get(/boom) assert response.status_code 500注意TestClient默认的raise_server_exceptionsTrue会让服务器端错误在测试用例中重新抛出要直接断言 500 响应需要传入raise_server_exceptionsFalse参见 tests/test_exceptions.py 的test_force_500_response。另外WebSocket 端点测试可结合client.websocket_connect()使用tests/test_exceptions.py。十、总结与最佳实践回顾本文的要点按状态码或异常类注册处理器exception_handlers映射的键既可以是int也可以是异常类异常类匹配基于 MRO 向上查找因此注册父类即可覆盖全部子类区分错误与可处理异常HTTPException代表可处理异常由ExceptionMiddleware转为响应其他异常是错误由最外层的ServerErrorMiddleware捕获为 500或交给500/Exception键的处理器并始终重新抛出供服务器记录日志debug 模式开启后错误会返回 HTML/纯文本 traceback 页面而不是 500 处理器仅用于开发环境切勿在生产开启HTTPException 与 WebSocketException分别用于 HTTP 端点与 WebSocket 端点HTTPException支持headers透传WebSocketException携带 RFC 6455 关闭码中间件编排ServerErrorMiddleware最外→ 用户中间件 →ExceptionMiddleware最内→ Router → Endpoints理解这一分层有助于判断异常究竟被谁捕获后台任务异常BackgroundTask抛出的异常会被错误处理器捕获但响应已发出新生成的响应会被丢弃因此应通过日志监控手段处理。这些行为全部有源码与测试背书可进一步阅读 starlette/middleware/exceptions.py、starlette/middleware/errors.py、starlette/_exception_handler.py 与 tests/test_exceptions.py 深化理解。【免费下载链接】starletteThe little ASGI framework that shines. 项目地址: https://gitcode.com/gh_mirrors/st/starlette创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考