sift_py.rest

 1from abc import ABC
 2from typing import TypedDict
 3
 4import requests
 5from requests.adapters import HTTPAdapter
 6from typing_extensions import NotRequired
 7from urllib3.util import Retry
 8
 9from sift_py.grpc.transport import _clean_uri
10
11_DEFAULT_REST_RETRY = Retry(total=3, status_forcelist=[500, 502, 503, 504], backoff_factor=1)
12
13
14class SiftRestConfig(TypedDict):
15    """
16    Config class used to to interact with services that use Sift's REST API.`.
17    - `uri`: The URI of Sift's REST API. The scheme portion of the URI i.e. `https://` should be ommitted.
18    - `apikey`: User-generated API key generated via the Sift application.
19    - `retry`: Urllib3 Retry configuration. If not provided, a default of 3 retries is used.
20    - `use_ssl`: INTERNAL USE. Meant to be used for local development.
21    - `cert_via_openssl`: Enable this if you want to use OpenSSL to load the certificates.
22    Run `pip install sift-stack-py[openssl]` to install the dependencies required to use this option.
23    Default is False.
24    """
25
26    uri: str
27    apikey: str
28    retry: NotRequired[Retry]
29    use_ssl: NotRequired[bool]
30    cert_via_openssl: NotRequired[bool]
31
32
33def compute_uri(restconf: SiftRestConfig) -> str:
34    uri = restconf["uri"]
35    use_ssl = restconf.get("use_ssl", True)
36    clean_uri = _clean_uri(uri, use_ssl)
37
38    if use_ssl:
39        return f"https://{clean_uri}"
40
41    return f"http://{clean_uri}"
42
43
44class _SiftHTTPAdapter(HTTPAdapter):
45    """Sift specific HTTP adapter."""
46
47    def __init__(self, rest_conf: SiftRestConfig, *args, **kwargs):
48        self._rest_conf = rest_conf
49        kwargs["max_retries"] = rest_conf.get("retry", _DEFAULT_REST_RETRY)
50        super().__init__(*args, **kwargs)
51
52    def init_poolmanager(self, *args, **kwargs):
53        if self._rest_conf.get("cert_via_openssl", False):
54            try:
55                import ssl
56
57                context = ssl.create_default_context()
58                context.load_default_certs()
59                kwargs["ssl_context"] = context
60            except ImportError as e:
61                raise Exception(
62                    "Missing required dependencies for cert_via_openssl. Run `pip install sift-stack-py[openssl]` to install the required dependencies."
63                ) from e
64        return super().init_poolmanager(*args, **kwargs)
65
66
67class _RestService(ABC):
68    """
69    Abstract service that implements a REST session.
70    """
71
72    def __init__(self, rest_conf: SiftRestConfig):
73        self._rest_conf = rest_conf
74        self._base_uri = compute_uri(rest_conf)
75        self._apikey = rest_conf["apikey"]
76
77        self._session = requests.Session()
78        self._session.headers = {"Authorization": f"Bearer {self._apikey}"}
79
80        adapter = _SiftHTTPAdapter(rest_conf)
81        self._session.mount("https://", adapter)
82        self._session.mount("http://", adapter)