97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
from config import LOGGER, NotAuthenticated
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.exceptions import HTTPException, StarletteHTTPException
|
|
from slowapi.errors import RateLimitExceeded
|
|
|
|
|
|
async def rate_limited(request: Request, error: RateLimitExceeded) -> JSONResponse:
|
|
n = 0
|
|
while True:
|
|
if error.detail[n].isnumeric():
|
|
n += 1
|
|
else:
|
|
break
|
|
error.detail = error.detail[:n] + " requests" + error.detail[n:]
|
|
LOGGER.info(
|
|
f"{request.client.host} got rate-limited at {str(request.url)} "
|
|
f"(exceeded {error.detail})"
|
|
)
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": f"Too many requests, retry after {error.detail[error.detail.find('per') + 4:]}",
|
|
"status_code": 429,
|
|
},
|
|
)
|
|
|
|
|
|
def not_authenticated(request: Request, _: NotAuthenticated) -> JSONResponse:
|
|
LOGGER.info(f"{request.client.host} failed to authenticate at {str(request.url)}")
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": "Authentication is required",
|
|
"status_code": 401,
|
|
},
|
|
)
|
|
|
|
|
|
def request_invalid(request: Request, exc: RequestValidationError) -> JSONResponse:
|
|
LOGGER.info(
|
|
f"{request.client.host} sent an invalid request at {request.url!r}: {type(exc).__name__}: {exc}"
|
|
)
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": f"Bad request: {type(exc).__name__}: {exc}",
|
|
"status_code": 400,
|
|
},
|
|
)
|
|
|
|
|
|
def http_exception(
|
|
request: Request, exc: HTTPException | StarletteHTTPException
|
|
) -> JSONResponse:
|
|
if exc.status_code >= 500:
|
|
LOGGER.error(
|
|
f"{request.client.host} raised a {exc.status_code} error at {request.url!r}:"
|
|
f"{type(exc).__name__}: {exc}"
|
|
)
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": "Internal server error",
|
|
"status_code": exc.status_code,
|
|
},
|
|
)
|
|
else:
|
|
LOGGER.info(
|
|
f"{request.client.host} raised an HTTP error ({exc.status_code}) at {str(request.url)}"
|
|
)
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": exc.detail,
|
|
"status_code": exc.status_code,
|
|
},
|
|
)
|
|
|
|
|
|
async def generic_error(request: Request, exc: Exception) -> JSONResponse:
|
|
"""
|
|
Handles generic, unexpected errors in the application
|
|
"""
|
|
|
|
LOGGER.info(
|
|
f"{request.client.host} raised an unexpected error ({type(exc).__name__}: {exc}) at {str(request.url)}"
|
|
)
|
|
return JSONResponse(
|
|
status_code=200,
|
|
content={
|
|
"msg": "Internal Server Error", # We can't leak anything about the error, it would be too risky
|
|
"status_code": 500,
|
|
},
|
|
)
|