93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
from typing import Any
|
|
|
|
from fastapi import FastAPI, HTTPException, Request, status
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
from src.utils.helpers import first
|
|
|
|
|
|
def register_exception_handler(app: FastAPI):
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
"""请求参数验证错误处理"""
|
|
error = first(exc.errors())
|
|
field = error.get('loc')[1] or ''
|
|
return JSONResponse(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
content=jsonable_encoder({
|
|
"status_code": status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
"message": "{} param {}".format(field, error.get('msg')),
|
|
"detail": exc.errors()}),
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
|
|
"""Http 异常处理"""
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=jsonable_encoder({
|
|
"status_code": exc.status_code,
|
|
"message": exc.detail}),
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def exception_callback(request: Request, exc: Exception):
|
|
|
|
"""其他异常处理，遇到其他问题再自定义异常"""
|
|
return JSONResponse(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
content=jsonable_encoder({
|
|
"status_code": status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
"message": "Internal Server Error",
|
|
# "detail": ''.join(exc.args)
|
|
}),
|
|
)
|
|
|
|
|
|
class CommonHttpException(HTTPException):
|
|
def __init__(self, detail, status_code, **kwargs: dict[str, Any]) -> None:
|
|
super().__init__(status_code=status_code, detail=detail, **kwargs)
|
|
|
|
|
|
class DetailedHTTPException(HTTPException):
|
|
STATUS_CODE = status.HTTP_500_INTERNAL_SERVER_ERROR
|
|
DETAIL = "Server error"
|
|
|
|
def __init__(self, **kwargs: dict[str, Any]) -> None:
|
|
super().__init__(status_code=self.STATUS_CODE, detail=self.DETAIL, **kwargs)
|
|
|
|
|
|
class PermissionDenied(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_403_FORBIDDEN
|
|
DETAIL = "Permission denied"
|
|
|
|
|
|
class NotFound(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_404_NOT_FOUND
|
|
|
|
|
|
class BadRequest(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_400_BAD_REQUEST
|
|
DETAIL = "Bad Request"
|
|
|
|
|
|
class UnprocessableEntity(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_422_UNPROCESSABLE_ENTITY
|
|
DETAIL = "Unprocessable entity"
|
|
|
|
|
|
class NotAuthenticated(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_401_UNAUTHORIZED
|
|
DETAIL = "User not authenticated"
|
|
|
|
|
|
class WxResponseError(DetailedHTTPException):
|
|
STATUS_CODE = status.HTTP_400_BAD_REQUEST
|
|
DETAIL = "请求微信异常"
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(headers={"WWW-Authenticate": "Bearer"})
|