| from datetime import datetime |
| from typing import Any, Optional |
|
|
| from pydantic import AnyHttpUrl, BaseModel, Extra, Field |
|
|
|
|
| class ExceptionReturn(BaseModel): |
| url: Optional[AnyHttpUrl] = None |
| time: datetime = Field(default_factory=datetime.now) |
| code: int = Field(ge=400, le=599) |
| detail: str |
| headers: dict[str, str] = {} |
|
|
| class Config: |
| extra = Extra.allow |
|
|
|
|
| class BaseServerException(Exception): |
| code: int = 500 |
| detail: str = "Server Fault" |
| headers: dict[str, Any] = {} |
|
|
| def __init__( |
| self, |
| detail: Optional[str] = None, |
| *, |
| code: Optional[int] = None, |
| headers: Optional[dict[str, Any]] = None, |
| **params |
| ) -> None: |
| self.data = ExceptionReturn( |
| detail=detail or self.__class__.detail, |
| code=code or self.__class__.code, |
| headers=headers or self.__class__.headers, |
| **params |
| ) |
| super().__init__(detail) |
|
|
|
|
| class BaseHTTPException(BaseServerException): |
| pass |
|
|
|
|
| class ServerSideException(BaseServerException): |
| code = 500 |
| detail = "Internal Server Error" |
|
|
|
|
| class UpstreamAPIException(ServerSideException): |
| code = 502 |
| detail = "Upstram API request failed" |
|
|
|
|
| class UncaughtException(ServerSideException): |
| code = 500 |
| detail = "Uncaught exception raised during processing" |
| exc: Exception |
|
|
| @classmethod |
| def with_exception(cls, e: Exception): |
| c = cls(e.__class__.__qualname__) |
| c.exc = e |
| return c |
|
|
|
|
| class ClientSideException(BaseServerException): |
| code = 400 |
| detail = "Bad Request" |
|
|
|
|
| class ValidationException(ClientSideException): |
| code = 422 |
|
|
|
|
| class RateLimitReachedException(ClientSideException): |
| code = 429 |
| detail = "Rate limit reached" |
|
|