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
54    def __attrs_post_init__(self):
55        self._headers["User-Agent"] = _get_user_agent(self.package_name, self.client_name)

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') 
 2def __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):
 3    self._base_url = base_url
 4    if cookies is not NOTHING:
 5        self._cookies = cookies
 6    else:
 7        self._cookies = __attr_factory__cookies()
 8    if headers is not NOTHING:
 9        self._headers = headers
10    else:
11        self._headers = __attr_factory__headers()
12    self._timeout = timeout
13    self._verify_ssl = verify_ssl
14    self._follow_redirects = follow_redirects
15    if httpx_args is not NOTHING:
16        self._httpx_args = httpx_args
17    else:
18        self._httpx_args = __attr_factory__httpx_args()
19    self._client = attr_dict['_client'].default
20    self._async_client = attr_dict['_async_client'].default
21    self.auth_method = auth_method
22    self.raise_on_unexpected_status = raise_on_unexpected_status
23    self.client_name = client_name
24    self.package_name = package_name
25    self.__attrs_post_init__()

Method generated by attrs for class CirroApiClient.

raise_on_unexpected_status: bool
client_name: str
package_name: 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) 
2def __init__(self, token):
3    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.

Inherited Members
httpx.Auth
requires_request_body
requires_response_body
sync_auth_flow
async_auth_flow
@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]) 
2def __init__(self, token_getter):
3    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.

Inherited Members
httpx.Auth
requires_request_body
requires_response_body
sync_auth_flow
async_auth_flow