exceptions.py 2.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293
  1. # (c) Nelen & Schuurmans
  2. from typing import Any
  3. from typing import List
  4. from typing import Optional
  5. from typing import Union
  6. from pydantic import create_model
  7. from pydantic import ValidationError
  8. from pydantic_core import ErrorDetails
  9. from .types import Id
  10. __all__ = [
  11. "AlreadyExists",
  12. "Conflict",
  13. "DoesNotExist",
  14. "PermissionDenied",
  15. "PreconditionFailed",
  16. "BadRequest",
  17. "Unauthorized",
  18. "BadRequest",
  19. ]
  20. class DoesNotExist(Exception):
  21. def __init__(self, name: str, id: Optional[Id] = None):
  22. super().__init__()
  23. self.name = name
  24. self.id = id
  25. def __str__(self):
  26. if self.id:
  27. return f"does not exist: {self.name} with id={self.id}"
  28. else:
  29. return f"does not exist: {self.name}"
  30. class Conflict(Exception):
  31. def __init__(self, msg: Optional[str] = None):
  32. super().__init__(msg)
  33. class AlreadyExists(Exception):
  34. def __init__(self, value: Any = None, key: str = "id"):
  35. super().__init__(f"record with {key}={value} already exists")
  36. class PreconditionFailed(Exception):
  37. def __init__(self, msg: str = "precondition failed", obj: Any = None):
  38. super().__init__(msg)
  39. self.obj = obj
  40. # pydantic.ValidationError needs some model; for us it doesn't matter
  41. # We do it the same way as FastAPI does it.
  42. request_model = create_model("Request")
  43. class BadRequest(Exception):
  44. def __init__(self, err_or_msg: Union[ValidationError, str]):
  45. self._internal_error = err_or_msg
  46. super().__init__(err_or_msg)
  47. def errors(self) -> List[ErrorDetails]:
  48. if isinstance(self._internal_error, ValidationError):
  49. return self._internal_error.errors()
  50. return [
  51. ErrorDetails(
  52. type="value_error",
  53. msg=self._internal_error,
  54. loc=[], # type: ignore
  55. input=None,
  56. )
  57. ]
  58. def __str__(self) -> str:
  59. error = self._internal_error
  60. if isinstance(error, ValidationError):
  61. details = error.errors()[0]
  62. loc = "'" + ",".join([str(x) for x in details["loc"]]) + "' "
  63. if loc == "'*' ":
  64. loc = ""
  65. return f"validation error: {loc}{details['msg']}"
  66. return f"validation error: {super().__str__()}"
  67. class Unauthorized(Exception):
  68. pass
  69. class PermissionDenied(Exception):
  70. pass