| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115 | import json as json_libimport refrom http import HTTPStatusfrom typing import Callablefrom typing import Optionalfrom urllib.parse import urlencodefrom urllib.parse import urljoinfrom pydantic import AnyHttpUrlfrom urllib3 import PoolManagerfrom urllib3 import Retryfrom clean_python import ctxfrom clean_python import Jsonfrom .exceptions import ApiException__all__ = ["SyncApiProvider"]def is_success(status: HTTPStatus) -> bool:    """Returns True on 2xx status"""    return (int(status) // 100) == 2JSON_CONTENT_TYPE_REGEX = re.compile(r"^application\/[^+]*[+]?(json);?.*$")def is_json_content_type(content_type: Optional[str]) -> bool:    if not content_type:        return False    return bool(JSON_CONTENT_TYPE_REGEX.match(content_type))def join(url: str, path: str) -> str:    """Results in a full url without trailing slash"""    assert url.endswith("/")    assert not path.startswith("/")    result = urljoin(url, path)    if result.endswith("/"):        result = result[:-1]    return resultdef add_query_params(url: str, params: Optional[Json]) -> str:    if params is None:        return url    return url + "?" + urlencode(params, doseq=True)class SyncApiProvider:    """Basic JSON API provider with retry policy and bearer tokens.    The default retry policy has 3 retries with 1, 2, 4 second intervals.    Args:        url: The url of the API (with trailing slash)        fetch_token: Callable that returns a token for a tenant id        retries: Total number of retries per request        backoff_factor: Multiplier for retry delay times (1, 2, 4, ...)    """    def __init__(        self,        url: AnyHttpUrl,        fetch_token: Callable[[PoolManager, int], Optional[str]],        retries: int = 3,        backoff_factor: float = 1.0,    ):        self._url = str(url)        assert self._url.endswith("/")        self._fetch_token = fetch_token        self._pool = PoolManager(retries=Retry(retries, backoff_factor=backoff_factor))    def request(        self,        method: str,        path: str,        params: Optional[Json] = None,        json: Optional[Json] = None,        fields: Optional[Json] = None,        timeout: float = 5.0,    ) -> Optional[Json]:        assert ctx.tenant is not None        headers = {}        request_kwargs = {            "method": method,            "url": add_query_params(join(self._url, path), params),            "timeout": timeout,        }        # for urllib3<2, we dump json ourselves        if json is not None and fields is not None:            raise ValueError("Cannot both specify 'json' and 'fields'")        elif json is not None:            request_kwargs["body"] = json_lib.dumps(json).encode()            headers["Content-Type"] = "application/json"        elif fields is not None:            request_kwargs["fields"] = fields        token = self._fetch_token(self._pool, ctx.tenant.id)        if token is not None:            headers["Authorization"] = f"Bearer {token}"        response = self._pool.request(headers=headers, **request_kwargs)        status = HTTPStatus(response.status)        content_type = response.headers.get("Content-Type")        if status is HTTPStatus.NO_CONTENT:            return None        if not is_json_content_type(content_type):            raise ApiException(                f"Unexpected content type '{content_type}'", status=status            )        body = json_lib.loads(response.data.decode())        if is_success(status):            return body        else:            raise ApiException(body, status=status)
 |