FastAPI基础入门(四)

FastAPI 基础入门

fastapi第四章

---

---

本章源码

4.异常以及错误

通常来说，Error（错误）表示应用程序存在较为严重的问题，可能会导致程序崩溃或无法正常运行。而Exception（异常）则通常表示应用程序在运行过程中出现了可预测的和可恢复的问题，这些问题可以被捕获和处理，应用程序可以继续正常运行。对异常有效跟踪可以明确具体的异常问题点，比如什么类型的异常被抛出、具体异常发生在什么位置、这个异常是由于什么原因被抛出的，这样不仅可以让应用程序更加健壮、易于调试，还可以帮助定位和处理问题。

在FastAPI框架中，所有的错误和异常都是Exception类的子类。在FastAPI框架的异常处理中常用的是HTTPException类，它用于处理HTTP状态码异常。还有一个ValueError类，它常处理值错误。FastAPI框架中的HTTPException类是基于Starlette框架的HTTPException类进行扩展的，以便支持自定义响应头。此外，FastAPI还引入了RequestValidationError子类来处理请求验证错误。

需要补充的是，FastAPI还提供了许多其他类型的异常类，如WebSocketError、WebSocketDisconnect和HTTPException 400等。这些异常类的使用方式与HTTPException和RequestValidationError类似，都是在路由处理函数中通过raise关键字抛出。FastAPI还允许用户自定义异常类，并提供了一些内置的异常处理器来处理这些自定义异常。

4.1 HTTPException异常类

通过名称可以知道，HTTPException异常和HTTP响应是存在关联的。HTTPException应用的场景主要是对客户端前端校验抛出指定HTTP响应状态异常。比如基于用户权限校验抛出403状态码，这表示当前客户端无访问权限；基于资源访问抛出404状态码，这表示找不到对应路由。这些异常通常都需要抛出，以告知客户端错误状态码及错误信息。
如果需要手动抛出异常，则需要使用raise关键字，而不能直接通过return返回异常。一般通过raise抛出异常之后，再对具体异常类型及异常信息进行分析，最后进行return处理。

4.1.1 HTTPException简单源码分析

HTTPException是基于StarletteHTTPException（它是HTTPException类的别名）实现的，这里新增了自定义的headers参数

from starlette.exceptions import HTTPException as StarletteHTTPException
class HTTPException(StarletteHTTPException):"""An HTTP exception you can raise in your own code to show errors to the client.This is for client errors, invalid authentication, invalid data, etc. Not for servererrors in your code.Read more about it in the[FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).## Example```pythonfrom fastapi import FastAPI, HTTPExceptionapp = FastAPI()items = {"foo": "The Foo Wrestlers"}@app.get("/items/{item_id}")async def read_item(item_id: str):if item_id not in items:raise HTTPException(status_code=404, detail="Item not found")return {"item": items[item_id]}def __init__(self,status_code: Annotated[int,Doc("""HTTP status code to send to the client."""),],detail: Annotated[Any,Doc("""Any data to be sent to the client in the `detail` key of the JSONresponse."""),] = None,headers: Annotated[Optional[Dict[str, str]],Doc("""Any headers to send to the client in the response."""),] = None,) -> None:super().__init__(status_code=status_code, detail=detail, headers=headers)class HTTPException(Exception):def __init__(self,status_code:int,detail:str = None)->None:if detail is None:detail = http.HTTPStatus(status_code).phraseself.status_code = status_codeself.detail = detaildef __repr__(self) -> str:class_name = self.__class__.__name__return f"{class_name}(status_code={self.status_code}, detail={self.detail!r})"

HTTPException的本质就是Exception的子类。HTTPException的参数项信息主要包含了以下的几部分：
❑ detail表示异常信息详细描述，它支持Any类型，所以它的值既可以是lits，也可以是dict，还可以是字符串等。
❑ status_code表示异常HTTP状态码值。
❑ headers表示响应报文头信息，这是FastAPI新增的部分。

4.1.2 HTTPException异常的使用

from fastapi import Request
from fastapi import FastAPI, Query, HTTPException
from starlette.responses import JSONResponse
app = FastAPI()
@app.get("/http_exception")
async def http_exception(action_scopes: str = Query(default='admin')):if action_scopes == 'admin':raise HTTPException(status_code=403,headers={"x-auth": "NO AUTH!"},detail={'code': '403','message': '错误当前你没有权限访问',})return {'code': '200', }

❑ 通过app.get()装饰器定义了一个路由，且路由地址为“/http_exception”。
❑ 通过app.get()装饰器绑定了一个名为http_exception的视图函数，在http_exception()视图函数中声明了一个action_scopes查询参数，其默认值为“admin”。
❑ 当访问路由时，若默认没有提交action_scopes查询参数值，则在视图函数内部会直接抛出raise HTTPException异常，并且在响应报文头中写入x-auth头信息。

启动服务并通过浏览器访问http://127.0.0.1:5000/http_exception/

4.1.3 覆盖HTTPException异常处理

全局异常拦截的方式覆盖并重写raise HTTPException抛出的异常

import asynciofrom fastapi import Request
from fastapi import FastAPI, Query, HTTPException
from starlette.responses import JSONResponse# 需要注意设置使用的事件循环方式#asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())app = FastAPI()@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):return JSONResponse(status_code=exc.status_code, content=exc.detail, headers=exc.headers)@app.get("/http_exception")
async def http_exception(action_scopes: str = Query(default='admin')):if action_scopes == 'admin':raise HTTPException(status_code=403,headers={"x-auth": "NO AUTH!"},detail={'code': '403','message': '错误当前你没有权限访问',})return {'code': '200', }

通过@app.exception_handler(HTTPException)进行一个全局异常的捕获处理。当手动抛出HTTPException异常时，FastAPI会对此类异常进行拦截处理，并进入http_exception_handler()函数。在该函数内部会对异常信息进行提取处理，然后通过JSONResponse进行响应报文封装输出并返回。启动服务，并通过浏览器访问http://127.0.0.1:5000/http_exception/

4.2 RequestValidationError错误

RequestValidationError主要是因为校验客户端提交的Request参数不合规才会抛出的错误。比如：
❑ 参数类型不符。
❑ 参数值长度不符。
❑ 参数格式不符。

4.2.1 RequestValidationError的使用

对Body、From、Path、Query等参数进行解析读取时，如果参数不符合要求，则会抛出RequestValidationError错误。

from fastapi import FastAPI
app = FastAPI()@app.get("/request_exception/")
async def request_exception(user_id: int):return {"user_id": user_id}

请求http://127.0.0.1:8000/request_exception/?user_id=xiao，会返回错误

从返回结果可知：
❑ 错误位置是query参数user_id。
❑ 错误类型是integer。
❑ 错误描述信息中指出了当前user_id的值不是一个int类型的值。
对参数进行校验得到RequestValidationError错误主要是结合Pydantic来实现的。

4.2.2 覆盖RequestValidationError错误

对RequestValidationError错误进行覆盖拦截处理

from fastapi import FastAPI, WebSocket, WebSocketDisconnectapp = FastAPI()@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):return JSONResponse({'mes':'触发了RequestValidationError错误，，错误信息:%s ！'%(str(exc))})@app.get("/request_exception/")
async def request_exception(user_id: int):return {"user_id": user_id}

当程序运行到参数校验阶段时，如果传入的参数user_id不是int类型，那么就会抛出RequestValidationError错误，此时FastAPI会对此类错误进行拦截处理，并进入validation_exception_handler()函数，在函数内部对错误信息进行提取处理后，通过JSONResponse进行响应报文封装输出。

4.3 自定义异常

4.3.1 自定义异常的实现

自定义异常可以继承自Exception或者其他已经实现的Exception类的子类。

from fastapi import Request
from fastapi import FastAPI, Query, HTTPException
from starlette.responses import JSONResponseapp = FastAPI()class CustomException(Exception):def __init__(self, message: str):self.message = message@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):return JSONResponse(content={"message": exc.message}, )@app.get("/custom_exception")
async def read_unicorn(name: str = 'zhong'):if name == "zhong":raise CustomException(message='抛出自定义异常')return {"name": name}

自定义了一个CustomException异常，并且通过@app.exception_handler(CustomException)装饰器进行了全局CustomException异常的捕获处理。当程序抛出CustomException异常时，FastAPI会对此类异常进行拦截处理，并进入custom_exception_handler()函数，在函数内部对异常信息进行提取处理后，通过JSONResponse进行响应报文封装输出。
请求：http://127.0.0.1:8000/custom_exception/?name=xiao

4.3.2 自定义内部错误码和异常

参考微信支付API的设计，再结合实际需求来定制属于自己的内部错误码

{"return_code":"SUCCESS",""" SUCCESS/FAIL字段是通信标识，非交易标识，交易是否成功需要查看result_code """"return_msg":"OK", # 当return_code为FAIL返回错误原因"err_code":"SYSTEMERROR", # 当return_code为FAIL返回错误代码"err_code_des":"系统错误" # 当return_code为FAIL返回错误描述
}

设置内部错误码机制的步骤。
❑ 第一步：通过enum来枚举错误码及错误描述。在一些API设计规范中，通常的做法是给每一条产线都分配不同的错误码区间，通过划分区间来有效避免出现重复错误码。

from enum import Enum
class ExceptionEnum(Enum):SUCCESS = ('0000','OK')FAILED = ('9999','系统异常')USER_NO_DATA = ('10001','用户不存在')USER_REGIESTER_ERROR = ('10002','注册异常')PERMISSIONS_ERROR = ('2000','用户权限错误')

❑ 第二步：定义好对应的错误码区间后，根据错误码来自定义异常类。在自定义的异常类中初始声明err_code和err_code_des两个参数

class BusinessError(Exception):__slots__ = ['err_code', 'err_code_des']def __init__(self,  result: ExceptionEnum = None, err_code: str = "00000", err_code_des: str = ""):if result:self.err_code = result.value[0]self.err_code_des = err_code_des or result.value[1]else:self.err_code = err_codeself.err_code_des = err_code_dessuper().__init__(self)

❑ 第三步：主要对上面自定义的异常类BusinessError进行拦截、覆盖、重写，再对返回的错误信息进行提取，最后通过JSONResponse封装以便按固定的格式返回

@app.exception_handler(BusinessError)
async def custom_exception_handler(request: Request, exc: BusinessError):return JSONResponse(content={'return_code':'FAIL','return_msg':'参数错误','err_code': exc.err_code,'err_code_des': exc.err_code_des,})

❑ 第四步：定义好异常拦截处理后，为了验证异常，需要定义一个业务逻辑接口，在接口中主动抛出BusinessError

@app.get("/custom_exception")
async def custom_exception(name: str = 'zhong'):if name == "xiaozhong":raise BusinessError(ExceptionEnum.USER_NO_DATA)return {"name": name}

请求http://127.0.0.1:8000/custom_exception?name=xiaozhong

4.4 中间件抛出自定义异常

在中间件抛出异常的情况下，是无法通过全局异常类拦截自定义异常的。

from fastapi import Request
from fastapi import FastAPI, Query, HTTPException
from starlette.responses import JSONResponseapp = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):# 故意直接的抛出异常trraise CustomException(message='抛出自定义异常')response = await call_next(request)
return responseclass CustomException(Exception):def __init__(self, message: str):self.message = message@app.exception_handler(Exception)
async def custom_exception_handler(request: Request, exc: Exception):print("触发全局自定义Exception")if isinstance(exc,CustomException):print("触发全局自定义CustomException")return JSONResponse(content={"message": exc.message}, )@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):print("触发全局自定义CustomException")return JSONResponse(content={"message": exc.message}, )@app.get("/custom_exception")
async def read_unicorn(name: str = 'zhong'):return {"name": name}

自定义异常示例代码的基础上新增了一个中间件“add_process_time_header”注册。当有请求进入时，首先会进入add_process_time_header中间件，在这个中间件中没做任何条件判断，直接抛出自定义CustomException异常。此时，控制台会抛出异常信息，但是全局异常错误custom_exception_handler并没有捕获到自定义的CustomException异常。无法捕获的根本原因是FastAPI框架在底层中对所有任何类型的中间件抛出的异常，都统归到顶层的ServerErrorMiddleware中间件进行捕获，而在ServerErrorMiddleware中间件中捕获到的异常都以Exception的方式抛出。

class ServerErrorMiddleware:"""Handles returning 500 responses when a server error occurs.
If 'debug' is set, then traceback responses will be returned,
otherwise the designated 'handler' will be called.This middleware class should generally be used to wrap *everything*
else up, so that unhandled exceptions anywhere in the stack
always result in an appropriate 500 response.
"""def __init__(self,app: ASGIApp,handler: typing.Callable[[Request, Exception], typing.Any] | None = None,debug: bool = False,
) -> None:self.app = appself.handler = handlerself.debug = debugasync def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:if scope["type"] != "http":await self.app(scope, receive, send)returnresponse_started = Falseasync def _send(message: Message) -> None:nonlocal response_started, sendif message["type"] == "http.response.start":response_started = Trueawait send(message)try:await self.app(scope, receive, _send)except Exception as exc:request = Request(scope)if self.debug:# In debug mode, return traceback responses.response = self.debug_response(request, exc)elif self.handler is None:# Use our default 500 error handler.response = self.error_response(request, exc)else:# Use an installed 500 error handler.if is_async_callable(self.handler):response = await self.handler(request, exc)else:response = await run_in_threadpool(self.handler, request, exc)if not response_started:await response(scope, receive, send)# We always continue to raise the exception.# This allows servers to log the error, or allows test clients# to optionally raise the error within the test case.raise exc

@app.exception_handler(Exception)
async def custom_exception_handler(request: Request, exc: Exception):print("触发全局自定义Exception")if isinstance(exc,CustomException):print("触发全局自定义CustomException")return JSONResponse(content={"message": exc.message}, )

此时再访问API，就会发现在中间件中手动抛出的自定义异常类被“@app.exception_handler(Exception)”给截获了。