from http import HTTPStatus from typing import Any from aiohttp import web import config class APIError(Exception): """ Base APIError class """ detail: str = None code: str = None headers: dict = None status_code: int = None extra: Any = None service: str = None original_status_code: int = None original_response: Any = None def __init__( self, detail: str = None, code: str = None, status_code: int = None, extra: Any = None, headers: dict = None, service: str = None, original_status_code: 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 original_status_code and original_status_code in config.ORIGINAL_STATUS_CODES_TO_BYPASS: self.status_code = original_status_code elif status_code: self.status_code = 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) -> 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.__class__.__name__}>: {self.to_dict()}" 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) -> web.Response: return web.json_response(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 Unauthorized(APIError): """HTTP 401 Unauthorized.""" detail = "Unauthorized." code = "unauthorized" status_code = HTTPStatus.UNAUTHORIZED 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 class RemoteServerError(APIError): """HTTP 530 Remote server error.""" detail = "Remote server error." code = "remote_server_error" status_code = 530