etesync-server/etebase_fastapi/exceptions.py

119 lines
3.3 KiB
Python
Raw Normal View History

2020-12-23 21:29:08 +00:00
from fastapi import status
2020-12-25 11:21:20 +00:00
import typing as t
from pydantic import BaseModel
2020-12-23 21:29:08 +00:00
2020-12-25 10:36:06 +00:00
2020-12-28 07:51:27 +00:00
class HttpErrorField(BaseModel):
2020-12-25 11:21:20 +00:00
field: str
code: str
detail: str
2020-12-28 15:39:51 +00:00
class Config:
orm_mode = True
2020-12-25 11:21:20 +00:00
2020-12-28 07:51:27 +00:00
class HttpErrorOut(BaseModel):
2020-12-25 11:21:20 +00:00
code: str
detail: str
2020-12-28 07:51:27 +00:00
errors: t.Optional[t.List[HttpErrorField]]
2020-12-25 11:21:20 +00:00
2020-12-28 15:39:51 +00:00
class Config:
orm_mode = True
2020-12-25 11:21:20 +00:00
2020-12-23 21:29:08 +00:00
class CustomHttpException(Exception):
def __init__(self, code: str, detail: str, status_code: int = status.HTTP_400_BAD_REQUEST):
self.status_code = status_code
self.code = code
self.detail = detail
@property
def as_dict(self) -> dict:
return {"code": self.code, "detail": self.detail}
class AuthenticationFailed(CustomHttpException):
def __init__(
self,
code="authentication_failed",
detail: str = "Incorrect authentication credentials.",
status_code: int = status.HTTP_401_UNAUTHORIZED,
):
super().__init__(code=code, detail=detail, status_code=status_code)
class NotAuthenticated(CustomHttpException):
def __init__(
self,
code="not_authenticated",
detail: str = "Authentication credentials were not provided.",
status_code: int = status.HTTP_401_UNAUTHORIZED,
):
super().__init__(code=code, detail=detail, status_code=status_code)
class PermissionDenied(CustomHttpException):
def __init__(
self,
code="permission_denied",
detail: str = "You do not have permission to perform this action.",
status_code: int = status.HTTP_403_FORBIDDEN,
):
super().__init__(code=code, detail=detail, status_code=status_code)
2020-12-25 10:36:06 +00:00
2020-12-28 07:51:27 +00:00
class HttpError(CustomHttpException):
2020-12-25 11:21:20 +00:00
def __init__(
self,
code: str,
detail: str,
status_code: int = status.HTTP_400_BAD_REQUEST,
2020-12-28 07:51:27 +00:00
errors: t.Optional[t.List["HttpError"]] = None,
2020-12-25 11:21:20 +00:00
):
self.errors = errors
2020-12-25 10:36:06 +00:00
super().__init__(code=code, detail=detail, status_code=status_code)
2020-12-25 11:21:20 +00:00
@property
def as_dict(self) -> dict:
2020-12-28 07:51:27 +00:00
return HttpErrorOut(code=self.code, errors=self.errors, detail=self.detail).dict()
2020-12-25 11:21:20 +00:00
2020-12-25 10:36:06 +00:00
2020-12-28 15:39:51 +00:00
class ValidationError(HttpError):
def __init__(
self,
code: str,
detail: str,
status_code: int = status.HTTP_400_BAD_REQUEST,
errors: t.Optional[t.List["HttpError"]] = None,
field: t.Optional[str] = None,
):
self.field = field
super().__init__(code=code, detail=detail, errors=errors, status_code=status_code)
2020-12-28 07:51:27 +00:00
def flatten_errors(field_name, errors) -> t.List[HttpError]:
2020-12-25 10:36:06 +00:00
ret = []
if isinstance(errors, dict):
for error_key in errors:
error = errors[error_key]
ret.extend(flatten_errors("{}.{}".format(field_name, error_key), error))
else:
for error in errors:
if error.messages:
message = error.messages[0]
else:
message = str(error)
2020-12-25 11:21:20 +00:00
ret.append(dict(code=error.code, detail=message, field=field_name))
2020-12-25 10:36:06 +00:00
return ret
def transform_validation_error(prefix, err):
if hasattr(err, "error_dict"):
errors = flatten_errors(prefix, err.error_dict)
elif not hasattr(err, "message"):
errors = flatten_errors(prefix, err.error_list)
else:
2020-12-28 07:51:27 +00:00
raise HttpError(err.code, err.message)
raise HttpError(code="field_errors", detail="Field validations failed.", errors=errors)