关于fastapi处理，之前处理的方式比较不优雅，对于http错误和校验错误信息，都直接的返回对应的错误信息全部的。如果对错误处理经过加工的话 应该更加优雅一点，此处理方案来自于：fastapi_contir框架的处理。我对它基于自己的原来基础做了些小调整。
旧的错误的处理输出
最终优化后的输出
优化流程
1 定义返回的json_response
完整代码是：
from typing import Any, Dict, Optional
# 自定义返回的错误的响应体信息
# ORJSONResponse一依赖于：orjson
from fastapi.responses import JSONResponse
import time
from fastapi.encoders import jsonable_encoder
import json
import datetime
import decimal
import typing
class CJsonEncoder(json.JSONEncoder):
def default(self, obj):
if hasattr(obj, 'keys') and hasattr(obj, '__getitem__'):
return dict(obj)
if isinstance(obj, datetime.datetime):
return obj.strftime('%Y-%m-%d %H:%M:%S')
if isinstance(obj, datetime.date):
return obj.strftime('%Y-%m-%d')
if isinstance(obj, decimal.Decimal):
return float(obj)
if isinstance(obj, bytes):
return str(obj, encoding='utf-8')
return json.JSONEncoder.default(self, obj)
class ApiResponse(JSONResponse):
# 定义返回响应码--如果不指定的话则默认都是返回200
http_status_code = 200
# 默认成功
api_code = 0
# 默认Node.如果是必选的，去掉默认值即可
result: Optional[Dict[str, Any]] = None # 结果可以是{} 或 []
message = '成功'
success = True
timestamp = int(time.time() * 1000)
def __init__(self, success= None, http_status_code=None, api_code=None, result=None, message=None, **options):
if result:
self.result = result
if message:
self.message = message
if api_code:
self.api_code = api_code
if success != None:
self.success = success
if http_status_code:
self.http_status_code = http_status_code
# 返回内容体
body = dict(
message=self.message,
code=self.api_code,
success=self.success,
result=self.result,
timestamp=self.timestamp,
# 形如request="POST v1/client/register"
# request=request.method + ' ' + self.get_url_no_param()
)
# jsonable_encoder 处理不同字符串返回 比如时间戳 datatime类型的处理
# super(ApiResponse, self).__init__(status_code=self.http_status_code, content=jsonable_encoder(body), **options)
# jsonable_encoder 处理不同字符串返回 比如时间戳 datatime类型的处理---jsonable_encoder无法格式化
# super(ApiResponse, self).__init__(status_code=self.http_status_code, content=jsonable_encoder(body), **options)
if options:
body = {**body, **options}
[options.pop(key) for key in [itemkey for itemkey in options.keys() if itemkey not in ['content', 'status_code', 'headers', 'media_type', 'background']]]
super(ApiResponse, self).__init__(status_code=self.http_status_code, content=body, **options)
# 这个render会自动调用，如果这里需要特殊的处理的话，可以重写这个地方
def render(self, content: typing.Any) -> bytes:
return json.dumps(
content,
ensure_ascii=False,
allow_nan=False,
indent=None,
separators=(",", ":"),
cls=CJsonEncoder
).encode("utf-8")
class BadrequestException(ApiResponse):
http_status_code = 400
api_code = 10031
result = None # 结果可以是{} 或 []
message = '错误的请求'
success = False
class LimiterResException(ApiResponse):
http_status_code = 429
api_code = 429
result = None # 结果可以是{} 或 []
message = '访问的速度过快'
success = False
class ParameterException(ApiResponse):
http_status_code = 400
result = {}
message = '参数校验错误,请检查提交的参数信息'
api_code = 10031
success = False
class UnauthorizedException(ApiResponse):
http_status_code = 401
result = {}
message = '未经许可授权'
api_code = 10032
success = False
class ForbiddenException(ApiResponse):
http_status_code = 403
result = {}
message = '失败！当前访问没有权限，或操作的数据没权限!'
api_code = 10033
success = False
class NotfoundException(ApiResponse):
http_status_code = 404
result = {}
message = '访问地址不存在'
api_code = 10034
success = False
class MethodnotallowedException(ApiResponse):
http_status_code = 405
result = {}
message = '不允许使用此方法提交访问'
api_code = 10034
success = False
class OtherException(ApiResponse):
http_status_code = 800
result = {}
message = '未知的其他HTTPEOOER异常'
api_code = 10034
success = False
class InternalErrorException(ApiResponse):
http_status_code = 200
result = {}
message = '程序员哥哥睡眠不足，系统崩溃了！'
api_code = 200
success = False
class InvalidTokenException(ApiResponse):
http_status_code = 401
api_code = 401
message = '很久没操作，令牌失效'
success = False
class ExpiredTokenException(ApiResponse):
http_status_code = 422
message = '很久没操作，令牌过期'
api_code = 10050
success = False
class FileTooLargeException(ApiResponse):
http_status_code = 413
api_code = 413
result = None # 结果可以是{} 或 []
message = '文件体积过大'
class FileTooManyException(ApiResponse):
http_status_code = 413
message = '文件数量过多'
api_code = 10120
result = None # 结果可以是{} 或 []
class FileExtensionException(ApiResponse):
http_status_code = 401
message = '文件扩展名不符合规范'
api_code = 10121
result = None # 结果可以是{} 或 []
class Success(ApiResponse):
http_status_code = 200
api_code = 200
result = None # 结果可以是{} 或 []
message = '自定义成功返回'
success = True
class Fail(ApiResponse):
http_status_code = 200
api_code = 200
result = None # 结果可以是{} 或 []
message = '自定义成功返回'
success = False
PS：在上面中对于如果有其他非restur参数的需要在外层传入的话
需要添加：
if options:
body = {**body, **options}
[options.pop(key) for key in [itemkey for itemkey in options.keys() if itemkey not in ['content', 'status_code', 'headers', 'media_type', 'background']]]
2 修改原来我的错误处理机制
原来的代码：
from fastapi import FastAPI, Request
from apps.response.json_response import *
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.exceptions import HTTPException as FastapiHTTPException
from fastapi.exceptions import RequestValidationError
from pydantic.errors import *
from apps.ext.logger import logger
import traceback
from apps.utils.singleton_helper import Singleton
@Singleton
class ApiExceptionHandler():
def __init__(self, app=None, *args, **kwargs):
super().__init__(*args, **kwargs)
if app is not None:
self.init_app(app)
def init_app(self, app: FastAPI):
# @app.exception_handler(StarletteHTTPException)
# @app.exception_handler(RequestValidationError)
# @app.exception_handler(Exception)
app.add_exception_handler(Exception, handler=self.all_exception_handler)
# 捕获StarletteHTTPException返回的错误异常，如返回405的异常的时候，走的是这个地方
app.add_exception_handler(StarletteHTTPException, handler=self.http_exception_handler)
app.add_exception_handler(RequestValidationError, handler=self.validation_exception_handler)
async def validation_exception_handler(self, request: Request, exc: RequestValidationError):
# print("参数提交异常错误selfself", exc.errors()[0].get('loc'))
# 路径参数错误
# 判断错误类型
if isinstance(exc.raw_errors[0].exc, IntegerError):
pass
elif isinstance(exc.raw_errors[0].exc, MissingError):
pass
return ParameterException(http_status_code=400, api_code=400, message='参数校验错误', result={
"detail": exc.errors(),
"body": exc.body
})
async def all_exception_handler(self, request: Request, exc: Exception):
'''
全局的捕获抛出的HTTPException异常，注意这里需要使用StarletteHTTPException的才可以
:param request:
:param exc:
:return:
'''
# log_msg = f"捕获到系统错误：请求路径:{request.url.path}\n错误信息：{traceback.format_exc()}"
if isinstance(exc, StarletteHTTPException) or isinstance(exc, FastapiHTTPException):
if exc.status_code == 405:
return MethodnotallowedException()
if exc.status_code == 404:
return NotfoundException()
elif exc.status_code == 429:
return LimiterResException()
elif exc.status_code == 500:
return InternalErrorException()
elif exc.status_code == 400:
# 有部分的地方直接的选择使用raise的方式抛出了异常，这里也需要进程处理
# raise HTTPException(HTTP_400_BAD_REQUEST, 'Invalid token')
return BadrequestException(msg=exc.detail)
return BadrequestException()
else:
# 其他内部的异常的错误拦截处理
logger.exception(exc)
traceback.print_exc()
return InternalErrorException()
async def http_exception_handler(self, request: Request, exc: StarletteHTTPException):
'''
全局的捕获抛出的HTTPException异常，注意这里需要使用StarletteHTTPException的才可以
:param request:
:param exc:
:return:
'''
# 这里全局监听了我们的所有的HTTP响应，包括了200 的也会尽到这里来！
# print("撒很好收到哈搜地和撒谎的撒22222222222===========",exc)
# log_msg = f"捕获到系统错误：请求路径:{request.url.path}\n错误信息：{traceback.format_exc()}"
if exc.status_code == 405:
return MethodnotallowedException()
if exc.status_code == 404:
return NotfoundException()
elif exc.status_code == 429:
return LimiterResException()
elif exc.status_code == 500:
return InternalErrorException()
elif exc.status_code == 400:
# 有部分的地方直接的选择使用raise的方式抛出了异常，这里也需要进程处理
# raise HTTPException(HTTP_400_BAD_REQUEST, 'Invalid token')
return BadrequestException(msg=exc.detail)
修改之后：
#!/usr/bin/evn python
# -*- coding: utf-8 -*-
from typing import Any, List, Optional
from fastapi import FastAPI, Request
from demo6.response.json_response import *
from starlette.exceptions import HTTPException as StarletteHTTPException
from fastapi.exceptions import HTTPException as FastapiHTTPException
from fastapi.exceptions import RequestValidationError
from pydantic.errors import *
import traceback
def Singleton(cls):
_instance = {}
def _singleton(*args, **kargs):
if cls not in _instance:
_instance[cls] = cls(*args, **kargs)
return _instance[cls]
return _singleton
def parse_error(err: Any, field_names: List, raw: bool = True) -> Optional[dict]:
"""
Parse single error object (such as pydantic-based or fastapi-based) to dict
:param err: Error object
:param field_names: List of names of the field that are already processed
:param raw: Whether this is a raw error or wrapped pydantic error
:return: dict with name of the field (or "__all__") and actual message
"""
if isinstance(err.exc, EnumError):
permitted_values = ", ".join([f"'{val}'" for val in err.exc.enum_values])
message = f"Value is not a valid enumeration member; "f"permitted: {permitted_values}."
elif isinstance(err.exc, StrRegexError):
message = "Provided value doesn't match valid format."
else:
message = str(err.exc) or ""
if not raw:
if len(err.loc_tuple()) == 2:
if str(err.loc_tuple()[0]) in ["body", "query"]:
name = err.loc_tuple()[1]
else:
name = err.loc_tuple()[0]
elif len(err.loc_tuple()) == 1:
if str(err.loc_tuple()[0]) == "body":
name = "__all__"
else:
name = str(err.loc_tuple()[0])
else:
name = "__all__"
else:
if len(err.loc_tuple()) == 2:
name = str(err.loc_tuple()[0])
elif len(err.loc_tuple()) == 1:
name = str(err.loc_tuple()[0])
else:
name = "__all__"
if name in field_names:
return None
if message and not any(
[message.endswith("."), message.endswith("?"), message.endswith("!")]
):
message = message + "."
message = message.capitalize()
return {"name": name, "message": message}
def raw_errors_to_fields(raw_errors: List) -> List[dict]:
"""
Translates list of raw errors (instances) into list of dicts with name/msg
:param raw_errors: List with instances of raw error
:return: List of dicts (1 dict for every raw error)
"""
fields = []
for top_err in raw_errors:
if hasattr(top_err.exc, "raw_errors"):
for err in top_err.exc.raw_errors:
# This is a special case when errors happen both in request
# handling & internal validation
if isinstance(err, list):
err = err[0]
field_err = parse_error(
err,
field_names=list(map(lambda x: x["name"], fields)),
raw=True,
)
if field_err is not None:
fields.append(field_err)
else:
field_err = parse_error(
top_err,
field_names=list(map(lambda x: x["name"], fields)),
raw=False,
)
if field_err is not None:
fields.append(field_err)
return fields
@Singleton
class ApiExceptionHandler():
def __init__(self, app=None, *args, **kwargs):
super().__init__(*args, **kwargs)
if app is not None:
self.init_app(app)
def init_app(self, app: FastAPI):
app.add_exception_handler(Exception, handler=self.all_exception_handler)
# 捕获StarletteHTTPException返回的错误异常，如返回405的异常的时候，走的是这个地方
app.add_exception_handler(StarletteHTTPException, handler=self.http_exception_handler)
app.add_exception_handler(RequestValidationError, handler=self.validation_exception_handler)
async def validation_exception_handler(self, request: Request, exc: RequestValidationError):
status_code = getattr(exc, "status_code", 400)
headers = getattr(exc, "headers", None)
fields = raw_errors_to_fields(exc.raw_errors)
message = getattr(exc, "message", "Validation error.")
if message and not any(
[message.endswith("."), message.endswith("?"), message.endswith("!")]
):
message = message + "." # pragma: no cover
data = {"message": message, "fields": fields}
return ParameterException(result=data)
async def all_exception_handler(self, request: Request, exc: Exception):
# log_msg = f"捕获到系统错误：请求路径:{request.url.path}\n错误信息：{traceback.format_exc()}"
if isinstance(exc, StarletteHTTPException) or isinstance(exc, FastapiHTTPException):
if exc.status_code == 405:
return MethodnotallowedException()
if exc.status_code == 404:
return NotfoundException()
elif exc.status_code == 429:
return LimiterResException()
elif exc.status_code == 500:
return InternalErrorException()
elif exc.status_code == 400:
# 有部分的地方直接的选择使用raise的方式抛出了异常，这里也需要进程处理
# raise HTTPException(HTTP_400_BAD_REQUEST, 'Invalid token')
return BadrequestException(msg=exc.detail)
return BadrequestException()
else:
traceback.print_exc()
return InternalErrorException()
async def http_exception_handler(self, request: Request, exc: StarletteHTTPException):
'''
全局的捕获抛出的HTTPException异常，注意这里需要使用StarletteHTTPException的才可以
:param request:
:param exc:
:return:
'''
if exc.status_code == 405:
return MethodnotallowedException()
elif exc.status_code == 404:
return NotfoundException()
elif exc.status_code == 429:
return LimiterResException()
elif exc.status_code == 500:
return InternalErrorException()
else:
fields = getattr(exc, "fields", [])
message = getattr(exc, "detail", "Validation error.")
if message and not any(
[message.endswith("."), message.endswith("?"), message.endswith("!")]
):
message = message + "."
data = {
"message": message,
"fields": fields,
}
return BadrequestException(result=data)
测试使用：
测试代码：
from fastapi import FastAPI
import uvicorn
from response.json_response import Success
from exception import ApiExceptionHandler
app = FastAPI()
@app.on_event("startup")
async def startup_event():
ApiExceptionHandler().init_app(app)
from typing import List
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str
description: str = Field(None, title="标题啊",description="错误提示文字啊", max_length=300)
price: float = Field(..., gt=0, description="错误提示文字啊")
tax: float = None
@app.get('/')
async def login(q: str = Query(..., min_length=3,max_length=50),
q2: str = Query(..., min_length=3,max_length=50),regex="^fixedquery$"):
return Success()
@app.get('/s')
async def index():
return Success()
if __name__ == '__main__':
# 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务：
# uvicorn xxx:app --reload debug=False 才会去走自己定义的异常信息
uvicorn.run('main:app',host="127.0.0.1", port=8000, debug=False, reload=True)
测试结果：
总结
以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！
结尾
END
简书：https://www.jianshu.com/u/d6960089b087
掘金：https://juejin.cn/user/2963939079225608
公众号：微信搜【小儿来一壶枸杞酒泡茶】
小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822
文章转载自小儿来一壶枸杞酒泡茶，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
