belvo.http

  1import logging
  2from typing import Dict, Generator, List, Union
  3
  4from requests import HTTPError, Session
  5
  6from belvo import __version__
  7from belvo.exceptions import RequestError
  8
  9logger = logging.getLogger(__name__)
 10
 11
 12class APISession:
 13    _secret_key_id: str
 14    _secret_key_password: str
 15    _url: str
 16
 17    def __init__(self, url: str) -> None:
 18        self._url = url
 19        self._session = Session()
 20        self._session.headers.update({"User-Agent": f"belvo-python ({__version__})"})
 21
 22    @property
 23    def url(self) -> Union[str, None]:
 24        return self._url
 25
 26    @property
 27    def key_id(self) -> Union[str, None]:
 28        return self._secret_key_id
 29
 30    @property
 31    def session(self) -> Session:
 32        return self._session
 33
 34    @property
 35    def headers(self) -> Dict:
 36        return self.session.headers  # type: ignore
 37
 38    def login(self, secret_key_id: str, secret_key_password: str, timeout: int = 5) -> bool:
 39        base_url = "{}/api/".format(self.url)
 40        self._secret_key_id = secret_key_id
 41        self._secret_key_password = secret_key_password
 42        self._session.auth = (secret_key_id, secret_key_password)
 43
 44        try:
 45            r = self.session.get(base_url, timeout=timeout)
 46            r.raise_for_status()
 47        except HTTPError:
 48            return False
 49        return True
 50
 51    def _get(self, url: str, params: Dict = None) -> Dict:
 52        if params is None:
 53            params = {}
 54
 55        timeout = params.pop("timeout", 5)
 56        r = self.session.get(url=url, params=params, timeout=timeout)
 57        r.raise_for_status()
 58        return r.json()
 59
 60    def get(self, endpoint: str, id: str, params: Dict = None) -> Dict:
 61        url = "{}{}{}/".format(self.url, endpoint, id)
 62        return self._get(url=url, params=params)
 63
 64    def put(
 65        self, endpoint: str, id: str, data: Dict, raise_exception: bool = False, **kwargs
 66    ) -> Union[List[Dict], Dict]:
 67        url = "{}{}{}/".format(self.url, endpoint, id)
 68        r = self.session.put(url=url, json=data, **kwargs)
 69
 70        if raise_exception:
 71            try:
 72                r.raise_for_status()
 73            except HTTPError:
 74                raise RequestError(r.status_code, r.json())
 75
 76        return r.json()
 77
 78    def list(self, endpoint: str, params: Dict = None) -> Generator:
 79        url = "{}{}".format(self.url, endpoint)
 80        while True:
 81            data = self._get(url, params=params)
 82            for result in data["results"]:
 83                yield result
 84
 85            if not data["next"]:
 86                break
 87
 88            url = data["next"]
 89            params = None
 90
 91    def post(
 92        self, endpoint: str, data: Dict, raise_exception: bool = False, *args, **kwargs
 93    ) -> Union[List, Dict]:
 94        url = "{}{}".format(self.url, endpoint)
 95        r = self.session.post(url, json=data, **kwargs)
 96
 97        if raise_exception:
 98            try:
 99                r.raise_for_status()
100            except HTTPError:
101                raise RequestError(r.status_code, r.json())
102
103        return r.json()
104
105    def patch(
106        self, endpoint: str, data: Dict, raise_exception: bool = False, **kwargs
107    ) -> Union[List[Dict], Dict]:
108        url = "{}{}".format(self.url, endpoint)
109        r = self.session.patch(url=url, json=data, **kwargs)
110
111        if raise_exception:
112            try:
113                r.raise_for_status()
114            except HTTPError:
115                raise RequestError(r.status_code, r.json())
116
117        return r.json()
118
119    def delete(self, endpoint: str, id: str, timeout: int = 5) -> bool:
120        url = "{}{}{}/".format(self.url, endpoint, id)
121        r = self.session.delete(url, timeout=timeout)
122        try:
123            r.raise_for_status()
124        except HTTPError:
125            return False
126        return True