cirro_api_client 

1from .cirro_auth import RefreshableTokenAuth, TokenAuth
2from .cirro_client import CirroApiClient
3
4__all__ = ("CirroApiClient", "TokenAuth", "RefreshableTokenAuth")
@define
class CirroApiClient(cirro_api_client.v1.client.Client): 
21@define
22class CirroApiClient(Client):
23    """A class for interacting with the Cirro API
24
25    Attributes:
26        auth_method: The method used to authenticate API requests
27
28        base_url: The base URL for the API, all requests are made to a relative path to this URL
29
30        cookies: A dictionary of cookies to be sent with every request
31
32        headers: A dictionary of headers to be sent with every request
33
34        timeout: The maximum amount of a time a request can take. API functions will raise
35        httpx.TimeoutException if this is exceeded.
36
37        verify_ssl: Whether to verify the SSL certificate of the API server. This should be True in production,
38        but can be set to False for testing purposes.
39
40        follow_redirects: Whether to follow redirects. Default value is False.
41
42        httpx_args: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor.
43
44        raise_on_unexpected_status: Whether to raise an errors.UnexpectedStatus if the API returns a
45                status code that was not documented in the source OpenAPI document. Can also be provided as a keyword
46                argument to the constructor.
47    """
48
49    auth_method: AuthMethod
50    raise_on_unexpected_status: bool = field(default=True, kw_only=True)
51    client_name: str = field(kw_only=True, default="Cirro API Client")
52    package_name: str = field(kw_only=True, default="cirro-api-client")
53    user_agent: str = field(init=False, default=None)
54
55    def __attrs_post_init__(self):
56        self.user_agent = _get_user_agent(self.package_name, self.client_name)
57        self._headers["User-Agent"] = self.user_agent

A class for interacting with the Cirro API

Attributes:
  • auth_method: The method used to authenticate API requests
  • base_url: The base URL for the API, all requests are made to a relative path to this URL
  • cookies: A dictionary of cookies to be sent with every request
  • headers: A dictionary of headers to be sent with every request
  • timeout: The maximum amount of a time a request can take. API functions will raise
  • httpx.TimeoutException if this is exceeded.
  • verify_ssl: Whether to verify the SSL certificate of the API server. This should be True in production,
  • but can be set to False for testing purposes.
  • follow_redirects: Whether to follow redirects. Default value is False.
  • httpx_args: A dictionary of additional arguments to be passed to the httpx.Client and httpx.AsyncClient constructor.
  • raise_on_unexpected_status: Whether to raise an errors.UnexpectedStatus if the API returns a status code that was not documented in the source OpenAPI document. Can also be provided as a keyword argument to the constructor.
CirroApiClient( base_url: str, auth_method: cirro_api_client.cirro_auth.AuthMethod, *, cookies: Dict[str, str] = NOTHING, headers: Dict[str, str] = NOTHING, timeout: Optional[httpx.Timeout] = None, verify_ssl: Union[str, bool, ssl.SSLContext] = True, follow_redirects: bool = False, httpx_args: Dict[str, Any] = NOTHING, raise_on_unexpected_status: bool = True, client_name: str = 'Cirro API Client', package_name: str = 'cirro-api-client') 
36def __init__(self, base_url, auth_method, *, cookies=NOTHING, headers=NOTHING, timeout=attr_dict['_timeout'].default, verify_ssl=attr_dict['_verify_ssl'].default, follow_redirects=attr_dict['_follow_redirects'].default, httpx_args=NOTHING, raise_on_unexpected_status=attr_dict['raise_on_unexpected_status'].default, client_name=attr_dict['client_name'].default, package_name=attr_dict['package_name'].default):
37    self._base_url = base_url
38    if cookies is not NOTHING:
39        self._cookies = cookies
40    else:
41        self._cookies = __attr_factory__cookies()
42    if headers is not NOTHING:
43        self._headers = headers
44    else:
45        self._headers = __attr_factory__headers()
46    self._timeout = timeout
47    self._verify_ssl = verify_ssl
48    self._follow_redirects = follow_redirects
49    if httpx_args is not NOTHING:
50        self._httpx_args = httpx_args
51    else:
52        self._httpx_args = __attr_factory__httpx_args()
53    self._client = attr_dict['_client'].default
54    self._async_client = attr_dict['_async_client'].default
55    self.auth_method = auth_method
56    self.raise_on_unexpected_status = raise_on_unexpected_status
57    self.client_name = client_name
58    self.package_name = package_name
59    self.user_agent = attr_dict['user_agent'].default
60    self.__attrs_post_init__()

Method generated by attrs for class CirroApiClient.

raise_on_unexpected_status: bool
client_name: str
package_name: str
user_agent: str
Inherited Members
cirro_api_client.v1.client.Client
get_auth
with_headers
with_cookies
with_timeout
set_httpx_client
get_httpx_client
set_async_httpx_client
get_async_httpx_client
@define
class TokenAuth(cirro_api_client.cirro_auth.AuthMethod): 
15@define
16class TokenAuth(AuthMethod):
17    token: str
18
19    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
20        request.headers["Authorization"] = f"Bearer {self.token}"
21        yield request

Defines the method used for authenticating with Cirro

TokenAuth(token: str) 
23def __init__(self, token):
24    self.token = token

Method generated by attrs for class TokenAuth.

token: str
def auth_flow( self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, NoneType]: 
19    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
20        request.headers["Authorization"] = f"Bearer {self.token}"
21        yield request

Execute the authentication flow.

To dispatch a request, yield it:

yield request

The client will .send() the response back into the flow generator. You can access it like so:

response = yield request

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.

@define
class RefreshableTokenAuth(cirro_api_client.cirro_auth.AuthMethod): 
24@define
25class RefreshableTokenAuth(AuthMethod):
26    token_getter: typing.Callable[[], str]
27
28    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
29        request.headers["Authorization"] = f"Bearer {self.token_getter()}"
30        yield request

Defines the method used for authenticating with Cirro

RefreshableTokenAuth(token_getter: Callable[[], str]) 
23def __init__(self, token_getter):
24    self.token_getter = token_getter

Method generated by attrs for class RefreshableTokenAuth.

token_getter: Callable[[], str]
def auth_flow( self, request: httpx.Request) -> Generator[httpx.Request, httpx.Response, NoneType]: 
28    def auth_flow(self, request: Request) -> typing.Generator[Request, Response, None]:
29        request.headers["Authorization"] = f"Bearer {self.token_getter()}"
30        yield request

Execute the authentication flow.

To dispatch a request, yield it:

yield request

The client will .send() the response back into the flow generator. You can access it like so:

response = yield request

A return (or reaching the end of the generator) will result in the client returning the last response obtained from the server.

You can dispatch as many requests as is necessary.