import hashlib import hmac import json import time from enum import IntEnum from http import HTTPStatus from typing import TYPE_CHECKING, Any, Optional import requests from ..._utils.json_decoder import JSONDecoder from ..._utils.json_encoder import JSONEncoder from ...exceptions import InvalidCredentialError from ..exceptions import NotFoundError, RequestParametersError, UnknownGenericError if TYPE_CHECKING: from requests.sessions import _Params class _Error(IntEnum): ERR_UNK = 10000 ERR_GENERIC = 10001 ERR_PARAMS = 10020 ERR_AUTH_FAIL = 10100 class Middleware: TIMEOUT = 30 def __init__( self, host: str, api_key: Optional[str] = None, api_secret: Optional[str] = None ): self.host, self.api_key, self.api_secret = host, api_key, api_secret def __build_authentication_headers(self, endpoint: str, data: Optional[str] = None): assert isinstance(self.api_key, str) and isinstance( self.api_secret, str ), "API_KEY and API_SECRET must be both strings" nonce = str(round(time.time() * 1_000_000)) if data is None: path = f"/api/v2/{endpoint}{nonce}" else: path = f"/api/v2/{endpoint}{nonce}{data}" signature = hmac.new( self.api_secret.encode("utf8"), path.encode("utf8"), hashlib.sha384 ).hexdigest() return { "bfx-nonce": nonce, "bfx-signature": signature, "bfx-apikey": self.api_key, } def _get(self, endpoint: str, params: Optional["_Params"] = None) -> Any: response = requests.get( url=f"{self.host}/{endpoint}", params=params, timeout=Middleware.TIMEOUT ) if response.status_code == HTTPStatus.NOT_FOUND: raise NotFoundError(f"No resources found at endpoint <{endpoint}>.") data = response.json(cls=JSONDecoder) if len(data) and data[0] == "error": if data[1] == _Error.ERR_PARAMS: raise RequestParametersError( "The request was rejected with the " f"following parameter error: <{data[2]}>" ) if ( data[1] is None or data[1] == _Error.ERR_UNK or data[1] == _Error.ERR_GENERIC ): raise UnknownGenericError( "The server replied to the request with " f"a generic error with message: <{data[2]}>." ) return data def _post( self, endpoint: str, params: Optional["_Params"] = None, body: Optional[Any] = None, _ignore_authentication_headers: bool = False, ) -> Any: data = body and json.dumps(body, cls=JSONEncoder) or None headers = {"Content-Type": "application/json"} if self.api_key and self.api_secret and not _ignore_authentication_headers: headers = {**headers, **self.__build_authentication_headers(endpoint, data)} response = requests.post( url=f"{self.host}/{endpoint}", params=params, data=data, headers=headers, timeout=Middleware.TIMEOUT, ) if response.status_code == HTTPStatus.NOT_FOUND: raise NotFoundError(f"No resources found at endpoint <{endpoint}>.") data = response.json(cls=JSONDecoder) if isinstance(data, list) and len(data) and data[0] == "error": if data[1] == _Error.ERR_PARAMS: raise RequestParametersError( "The request was rejected with the " f"following parameter error: <{data[2]}>" ) if data[1] == _Error.ERR_AUTH_FAIL: raise InvalidCredentialError( "Cannot authenticate with given API-KEY and API-SECRET." ) if ( data[1] is None or data[1] == _Error.ERR_UNK or data[1] == _Error.ERR_GENERIC ): raise UnknownGenericError( "The server replied to the request with " f"a generic error with message: <{data[2]}>." ) return data