cirro_api_client.v1.client

  1import ssl
  2from typing import Any, Dict, Optional, Union
  3
  4import httpx
  5from attrs import define, evolve, field
  6
  7
  8@define
  9class Client:
 10    """A class for keeping track of data related to the API
 11
 12    The following are accepted as keyword arguments and will be used to construct httpx Clients internally:
 13
 14        ``base_url``: The base URL for the API, all requests are made to a relative path to this URL
 15
 16        ``cookies``: A dictionary of cookies to be sent with every request
 17
 18        ``headers``: A dictionary of headers to be sent with every request
 19
 20        ``timeout``: The maximum amount of a time a request can take. API functions will raise
 21        httpx.TimeoutException if this is exceeded.
 22
 23        ``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production,
 24        but can be set to False for testing purposes.
 25
 26        ``follow_redirects``: Whether or not to follow redirects. Default value is False.
 27
 28        ``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
 29
 30
 31    Attributes:
 32        raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a
 33            status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
 34            argument to the constructor.
 35
 36        auth_method: Optionally provide a method used to authenticate API requests.
 37    """
 38
 39    raise_on_unexpected_status: bool = field(default=False, kw_only=True)
 40    auth_method: httpx.Auth = field(default=None, kw_only=True)
 41    _base_url: str
 42    _cookies: Dict[str, str] = field(factory=dict, kw_only=True)
 43    _headers: Dict[str, str] = field(factory=dict, kw_only=True)
 44    _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True)
 45    _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True)
 46    _follow_redirects: bool = field(default=False, kw_only=True)
 47    _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True)
 48    _client: Optional[httpx.Client] = field(default=None, init=False)
 49    _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False)
 50
 51    def get_auth(self):
 52        return self.auth_method
 53
 54    def with_headers(self, headers: Dict[str, str]) -> "Client":
 55        """Get a new client matching this one with additional headers"""
 56        if self._client is not None:
 57            self._client.headers.update(headers)
 58        if self._async_client is not None:
 59            self._async_client.headers.update(headers)
 60        return evolve(self, headers={**self._headers, **headers})
 61
 62    def with_cookies(self, cookies: Dict[str, str]) -> "Client":
 63        """Get a new client matching this one with additional cookies"""
 64        if self._client is not None:
 65            self._client.cookies.update(cookies)
 66        if self._async_client is not None:
 67            self._async_client.cookies.update(cookies)
 68        return evolve(self, cookies={**self._cookies, **cookies})
 69
 70    def with_timeout(self, timeout: httpx.Timeout) -> "Client":
 71        """Get a new client matching this one with a new timeout (in seconds)"""
 72        if self._client is not None:
 73            self._client.timeout = timeout
 74        if self._async_client is not None:
 75            self._async_client.timeout = timeout
 76        return evolve(self, timeout=timeout)
 77
 78    def set_httpx_client(self, client: httpx.Client) -> "Client":
 79        """Manually the underlying httpx.Client
 80
 81        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
 82        """
 83        self._client = client
 84        return self
 85
 86    def get_httpx_client(self) -> httpx.Client:
 87        """Get the underlying httpx.Client, constructing a new one if not previously set"""
 88        if self._client is None:
 89            self._client = httpx.Client(
 90                base_url=self._base_url,
 91                cookies=self._cookies,
 92                headers=self._headers,
 93                timeout=self._timeout,
 94                verify=self._verify_ssl,
 95                follow_redirects=self._follow_redirects,
 96                **self._httpx_args,
 97            )
 98        return self._client
 99
100    def __enter__(self) -> "Client":
101        """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
102        self.get_httpx_client().__enter__()
103        return self
104
105    def __exit__(self, *args: Any, **kwargs: Any) -> None:
106        """Exit a context manager for internal httpx.Client (see httpx docs)"""
107        self.get_httpx_client().__exit__(*args, **kwargs)
108
109    def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
110        """Manually the underlying httpx.AsyncClient
111
112        **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
113        """
114        self._async_client = async_client
115        return self
116
117    def get_async_httpx_client(self) -> httpx.AsyncClient:
118        """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
119        if self._async_client is None:
120            self._async_client = httpx.AsyncClient(
121                base_url=self._base_url,
122                cookies=self._cookies,
123                headers=self._headers,
124                timeout=self._timeout,
125                verify=self._verify_ssl,
126                follow_redirects=self._follow_redirects,
127                **self._httpx_args,
128            )
129        return self._async_client
130
131    async def __aenter__(self) -> "Client":
132        """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
133        await self.get_async_httpx_client().__aenter__()
134        return self
135
136    async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
137        """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
138        await self.get_async_httpx_client().__aexit__(*args, **kwargs)