error_responses.py 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. # (c) Nelen & Schuurmans
  2. from typing import List
  3. from typing import Union
  4. from fastapi.encoders import jsonable_encoder
  5. from fastapi.requests import Request
  6. from fastapi.responses import JSONResponse
  7. from starlette import status
  8. from clean_python import BadRequest
  9. from clean_python import Conflict
  10. from clean_python import DoesNotExist
  11. from clean_python import PermissionDenied
  12. from clean_python import Unauthorized
  13. from clean_python import ValueObject
  14. __all__ = [
  15. "ValidationErrorResponse",
  16. "DefaultErrorResponse",
  17. "not_found_handler",
  18. "conflict_handler",
  19. "validation_error_handler",
  20. "not_implemented_handler",
  21. "permission_denied_handler",
  22. "unauthorized_handler",
  23. ]
  24. class ValidationErrorEntry(ValueObject):
  25. loc: List[Union[str, int]]
  26. msg: str
  27. type: str
  28. class ValidationErrorResponse(ValueObject):
  29. detail: List[ValidationErrorEntry]
  30. class DefaultErrorResponse(ValueObject):
  31. message: str
  32. async def not_found_handler(request: Request, exc: DoesNotExist):
  33. return JSONResponse(
  34. status_code=status.HTTP_404_NOT_FOUND,
  35. content={"message": f"Could not find {exc.name} with id={exc.id}"},
  36. )
  37. async def conflict_handler(request: Request, exc: Conflict):
  38. return JSONResponse(
  39. status_code=status.HTTP_409_CONFLICT,
  40. content={"message": str(exc)},
  41. )
  42. async def validation_error_handler(request: Request, exc: BadRequest):
  43. return JSONResponse(
  44. status_code=status.HTTP_400_BAD_REQUEST,
  45. content=jsonable_encoder({"detail": exc.errors()}),
  46. )
  47. async def not_implemented_handler(request: Request, exc: NotImplementedError):
  48. return JSONResponse(
  49. status_code=status.HTTP_501_NOT_IMPLEMENTED,
  50. content={"message": str(exc)},
  51. )
  52. async def unauthorized_handler(request: Request, exc: Unauthorized):
  53. return JSONResponse(
  54. status_code=status.HTTP_401_UNAUTHORIZED,
  55. content={"message": "Unauthorized"},
  56. headers={"WWW-Authenticate": "Bearer"},
  57. )
  58. async def permission_denied_handler(request: Request, exc: PermissionDenied):
  59. return JSONResponse(
  60. status_code=status.HTTP_403_FORBIDDEN,
  61. content={"message": "Permission denied", "detail": str(exc)},
  62. )