""" HTTP API Client """ from typing import Dict, Any, Optional, TypeVar, Type, List, Union import aiohttp import backoff import marshmallow_dataclass from aiohttp import ClientResponse, ClientResponseError, ContentTypeError from colors import colors from marshmallow import ValidationError from dataclasses import dataclass, field import config import traceback from api import app from utils.exceptions import BadGateway, NotFound, APIError from external_api.base.clients.auth_strategies.Base import DefaultAuthStrategy, AuthStrategy from constants.http_status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN, HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND class BaseClientError(Exception): code: str detail: str def __init__(self, code: str, detail: str): self.code = code self.detail = detail class ApiUnavailable(BaseClientError): pass class ApiUnauthorized(BaseClientError): pass class ApiInvalidRequestError(BaseClientError): server_error: Optional[str] = None def __init__(self, error: Optional[str] = None): self.server_error = error class ApiResponseParsingError(BaseClientError): pass T = TypeVar("T") class ApiClient: _session: Optional[aiohttp.ClientSession] = None base_headers: Optional[Dict[str, str]] = None api_host: str auth_strategy: AuthStrategy = DefaultAuthStrategy() def __init__( self, api_host: str, auth_strategy: AuthStrategy = DefaultAuthStrategy(), headers: Optional[Dict[str, str]] = None, ): self.base_headers = headers self.api_host = api_host self.auth_strategy = auth_strategy @property def session(self): if self._session is None: self._session = aiohttp.ClientSession(headers=self.base_headers) return self._session async def get_list( self, endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, response_type: Type[T] = None, ) -> List[T]: return await self.get(endpoint, params, headers, response_type=response_type) async def get( self, endpoint: str, params: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, response_type: Type[T] = None, ) -> T: return await self.execute("get", endpoint=endpoint, params=params, headers=headers, response_type=response_type) async def post( self, endpoint: str, payload: Optional[Union[Dict[str, Any], List]] = None, headers: Optional[Dict[str, str]] = None, response_type: Type[T] = None, ) -> T: return await self.execute("post", endpoint=endpoint, json=payload, headers=headers, response_type=response_type) async def execute( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: Optional[Union[Dict[str, Any], List]] = None, headers: Optional[Dict[str, str]] = None, response_type: Type[T] = None, ) -> T: try: response = await self._execute(method, endpoint=endpoint, params=params, json=json, headers=headers) except ClientResponseError as error: self.log_request(endpoint, json, method, params, 502) raise BadGateway(detail="Code: {}. Message: {}".format(error.status, error.message)) except BaseClientError as error: app.logger.error("Execute http request Error: ", f"{error}\n{traceback.format_exc()}") raise BadGateway() except APIError as error: app.logger.error("Execute http request Error: ", f"{error}\n{traceback.format_exc()}") raise BadGateway() if 'application/json' not in response.headers.get('Content-Type'): return try: json_body = await response.json() except ContentTypeError as error: raise ApiResponseParsingError(code="validation_error", detail=error.message) if response_type is None: return json_body is_response_array = isinstance(json_body, list) schema = marshmallow_dataclass.class_schema(response_type)(many=is_response_array) try: data = schema.load(json_body) except ValidationError as map_error: raise ApiResponseParsingError(code="validation_error", detail=map_error.normalized_messages()) return data def log_request(self, endpoint, json, method, params, status): log_params = [ ("[{}] {}".format(method, endpoint), "blue"), ("Code: {}".format(status), "yellow"), ("Host: {}".format(self.api_host), "red"), ("Params: {}".format(params), "blue"), ("Body: {}".format(json), "blue"), ] if config.runtime_config().DEBUG: parts = [colors.color("{}".format(value), fg=color) for value, color in log_params] else: parts = [value for value, color in log_params] app.logger.info(" ".join(parts)) @backoff.on_exception(backoff.expo, ApiUnavailable, max_time=3) async def _execute( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: Optional[Union[Dict[str, Any], List]] = None, headers: Optional[Dict[str, str]] = None, ) -> Any: try: if headers is None: headers = {} headers = await self.auth_strategy.authorize(headers) response: ClientResponse = await self.session.request( method, self.api_host + endpoint, params=params, headers=headers, json=json ) self.log_request(endpoint, json, method, params, response.status) if response.status in (HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN): raise ApiUnauthorized(code="Unauthorized", detail="Token is expired or not valid") if response.status == HTTP_404_NOT_FOUND: raise NotFound() if response.status == HTTP_400_BAD_REQUEST: server_error = await response.text() raise ApiInvalidRequestError(server_error) return response except (aiohttp.ClientConnectorError, aiohttp.ServerTimeoutError): raise ApiUnavailable class FivetranApiClient(ApiClient): # async def execute( # self, # method: str, # endpoint: str, # params: Optional[Dict[str, Any]] = None, # json: Optional[Union[Dict[str, Any], List]] = None, # headers: Optional[Dict[str, str]] = None, # response_type: Type[T] = None, # ) -> T: # # try: # response = await self._execute(method, endpoint=endpoint, params=params, json=json, headers=headers) # except ClientResponseError as error: # self.log_request(endpoint, json, method, params, 502) # raise BadGateway(detail="Code: {}. Message: {}".format(error.status, error.message)) # except BaseClientError as error: # app.logger.error("Execute http request Error: ", f"{error}\n{traceback.format_exc()}") # raise BadGateway() # except APIError as error: # app.logger.error("Execute http request Error: ", f"{error}\n{traceback.format_exc()}") # raise BadGateway() # # if 'application/json' not in response.headers.get('Content-Type'): # return # # try: # json_body = await response.json() # except ContentTypeError as error: # raise ApiResponseParsingError(code="validation_error", detail=error.message) # # if response_type is None: # return json_body # # base_schema = marshmallow_dataclass.class_schema(FivetranBaseResponse) # try: # response = base_schema.load(json_body) # schema = marshmallow_dataclass.class_schema(response_type) # data = schema.load(**response) # except ValidationError as map_error: # raise ApiResponseParsingError(code="validation_error", detail=map_error.normalized_messages()) # # return data @backoff.on_exception(backoff.expo, ApiUnavailable, max_time=3) async def _execute( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: Optional[Union[Dict[str, Any], List]] = None, headers: Optional[Dict[str, str]] = None, ) -> Any: try: if headers is None: headers = { "Content-Type": "application/json", "Accept": "application/json;version=2" } basic_auth = self.auth_strategy.authorize({}) response: ClientResponse = await self.session.request( method, self.api_host + endpoint, params=params, headers=headers, json=json, auth=basic_auth ) self.log_request(endpoint, json, method, params, response.status) if response.status in (HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN): raise ApiUnauthorized(code="Unauthorized", detail="Id or secret not valid") if response.status == HTTP_404_NOT_FOUND: raise NotFound() if response.status == HTTP_400_BAD_REQUEST: server_error = await response.text() raise ApiInvalidRequestError(server_error) return response except (aiohttp.ClientConnectorError, aiohttp.ServerTimeoutError): raise ApiUnavailable