from http import HTTPStatus from typing import Any, Dict, Optional from requests import Response # type: ignore class APIError(Exception): """ Base APIError class """ detail: Optional[str] = None code: Optional[str] = None headers: Optional[Dict[Any, Any]] = None status_code: int = 200 extra: Any = None service: Optional[str] = None original_status_code: Optional[int] = None original_response: Any = None def __init__( self, detail: Optional[str] = None, code: Optional[str] = None, status_code: int = 200, extra: Any = None, headers: Optional[Dict[Any, Any]] = None, service: Optional[str] = None, original_status_code: Optional[int] = None, original_response: Any = None, ): """Init API error. Args: detail: Error detailed description. code: Error string code. status_code: Error status code. extra: Some extra data. headers: Error headers. service: Error from this service. original_status_code: Original other service error status code. original_response: Other service error body. """ if self.__class__ == APIError: raise NotImplementedError if status_code: self.status_code = status_code or self.status_code if detail: self.detail = detail if code: self.code = code if extra: self.extra = extra if headers: self.headers = headers if service: self.service = service if original_status_code: self.original_status_code = original_status_code if original_response: self.original_response = original_response def get_status_code(self) -> Optional[int]: """Get original error status code, use original_status_code if it is set else status_code. Returns: Status code. """ return self.original_status_code or self.status_code def __str__(self): return f"{self.status_code} {self.detail} {self.extra}" def to_dict(self): """ Serialize exception instance to a dictionary. Primary use case is to prepare exception for dumping it to JSON and adding to response body. Resulting dictionary always contains `code` and `detail` attributes. `extra` attribute is also added if it is not None. Returns: dict: Exception serialized to a dictionary. """ result = {"code": self.code, "detail": self.detail} if self.extra: result["extra"] = self.extra if self.service or self.original_status_code or self.original_response: result["original"] = { **({"service": self.service} if self.service else {}), **({"status_code": self.original_status_code} if self.original_status_code else {}), **({"response": self.original_response} if self.original_response else {}), } return result def respond(self) -> Response: return Response.json(data=self.to_dict(), status=self.status_code) class BadRequest(APIError): """HTTP 400 BadRequest.""" detail = "Bad Request" code = "bad_request" status_code = HTTPStatus.BAD_REQUEST class BadGateway(APIError): """ HTTP 502 Bad Gateway. This exception may be raised if a third party API returns invalid response. """ detail = "Bad Gateway." code = "bad_gateway" status_code = HTTPStatus.BAD_GATEWAY