Skip to content

katana_public_api_client.client

katana_public_api_client.client

Classes

AuthenticatedClient

A Client which has been authenticated for use on secured endpoints

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``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 or not 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 or not 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.
Functions
__aenter__() async

Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)

Source code in katana_public_api_client/client.py 
260
261
262
263
async def __aenter__(self) -> "AuthenticatedClient":
    """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
    await self.get_async_httpx_client().__aenter__()
    return self
__aexit__(*args, **kwargs) async

Exit a context manager for underlying httpx.AsyncClient (see httpx docs)

Source code in katana_public_api_client/client.py 
265
266
267
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
    await self.get_async_httpx_client().__aexit__(*args, **kwargs)
__enter__()

Enter a context manager for self.client—you cannot enter twice (see httpx docs)

Source code in katana_public_api_client/client.py 
224
225
226
227
def __enter__(self) -> "AuthenticatedClient":
    """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
    self.get_httpx_client().__enter__()
    return self
__exit__(*args, **kwargs)

Exit a context manager for internal httpx.Client (see httpx docs)

Source code in katana_public_api_client/client.py 
229
230
231
def __exit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for internal httpx.Client (see httpx docs)"""
    self.get_httpx_client().__exit__(*args, **kwargs)
get_async_httpx_client()

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in katana_public_api_client/client.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._headers[self.auth_header_name] = (
            f"{self.prefix} {self.token}" if self.prefix else self.token
        )
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client
get_httpx_client()

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in katana_public_api_client/client.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._headers[self.auth_header_name] = (
            f"{self.prefix} {self.token}" if self.prefix else self.token
        )
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client
set_async_httpx_client(async_client)

Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in katana_public_api_client/client.py 
233
234
235
236
237
238
239
240
241
def set_async_httpx_client(
    self, async_client: httpx.AsyncClient
) -> "AuthenticatedClient":
    """Manually the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self
set_httpx_client(client)

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in katana_public_api_client/client.py 
199
200
201
202
203
204
205
def set_httpx_client(self, client: httpx.Client) -> "AuthenticatedClient":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self
with_cookies(cookies)

Get a new client matching this one with additional cookies

Source code in katana_public_api_client/client.py 
183
184
185
186
187
188
189
def with_cookies(self, cookies: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})
with_headers(headers)

Get a new client matching this one with additional headers

Source code in katana_public_api_client/client.py 
175
176
177
178
179
180
181
def with_headers(self, headers: dict[str, str]) -> "AuthenticatedClient":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})
with_timeout(timeout)

Get a new client matching this one with a new timeout (in seconds)

Source code in katana_public_api_client/client.py 
191
192
193
194
195
196
197
def with_timeout(self, timeout: httpx.Timeout) -> "AuthenticatedClient":
    """Get a new client matching this one with a new timeout (in seconds)"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)

Client

A class for keeping track of data related to the API

The following are accepted as keyword arguments and will be used to construct httpx Clients internally:

``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 or not 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 or not 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.
Functions
__aenter__() async

Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)

Source code in katana_public_api_client/client.py 
123
124
125
126
async def __aenter__(self) -> "Client":
    """Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)"""
    await self.get_async_httpx_client().__aenter__()
    return self
__aexit__(*args, **kwargs) async

Exit a context manager for underlying httpx.AsyncClient (see httpx docs)

Source code in katana_public_api_client/client.py 
128
129
130
async def __aexit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for underlying httpx.AsyncClient (see httpx docs)"""
    await self.get_async_httpx_client().__aexit__(*args, **kwargs)
__enter__()

Enter a context manager for self.client—you cannot enter twice (see httpx docs)

Source code in katana_public_api_client/client.py 
92
93
94
95
def __enter__(self) -> "Client":
    """Enter a context manager for self.client—you cannot enter twice (see httpx docs)"""
    self.get_httpx_client().__enter__()
    return self
__exit__(*args, **kwargs)

Exit a context manager for internal httpx.Client (see httpx docs)

Source code in katana_public_api_client/client.py 
97
98
99
def __exit__(self, *args: Any, **kwargs: Any) -> None:
    """Exit a context manager for internal httpx.Client (see httpx docs)"""
    self.get_httpx_client().__exit__(*args, **kwargs)
get_async_httpx_client()

Get the underlying httpx.AsyncClient, constructing a new one if not previously set

Source code in katana_public_api_client/client.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
def get_async_httpx_client(self) -> httpx.AsyncClient:
    """Get the underlying httpx.AsyncClient, constructing a new one if not previously set"""
    if self._async_client is None:
        self._async_client = httpx.AsyncClient(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._async_client
get_httpx_client()

Get the underlying httpx.Client, constructing a new one if not previously set

Source code in katana_public_api_client/client.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_httpx_client(self) -> httpx.Client:
    """Get the underlying httpx.Client, constructing a new one if not previously set"""
    if self._client is None:
        self._client = httpx.Client(
            base_url=self._base_url,
            cookies=self._cookies,
            headers=self._headers,
            timeout=self._timeout,
            verify=self._verify_ssl,
            follow_redirects=self._follow_redirects,
            **self._httpx_args,
        )
    return self._client
set_async_httpx_client(async_client)

Manually the underlying httpx.AsyncClient

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in katana_public_api_client/client.py 
101
102
103
104
105
106
107
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "Client":
    """Manually the underlying httpx.AsyncClient

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._async_client = async_client
    return self
set_httpx_client(client)

Manually set the underlying httpx.Client

NOTE: This will override any other settings on the client, including cookies, headers, and timeout.

Source code in katana_public_api_client/client.py 
70
71
72
73
74
75
76
def set_httpx_client(self, client: httpx.Client) -> "Client":
    """Manually set the underlying httpx.Client

    **NOTE**: This will override any other settings on the client, including cookies, headers, and timeout.
    """
    self._client = client
    return self
with_cookies(cookies)

Get a new client matching this one with additional cookies

Source code in katana_public_api_client/client.py 
54
55
56
57
58
59
60
def with_cookies(self, cookies: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional cookies"""
    if self._client is not None:
        self._client.cookies.update(cookies)
    if self._async_client is not None:
        self._async_client.cookies.update(cookies)
    return evolve(self, cookies={**self._cookies, **cookies})
with_headers(headers)

Get a new client matching this one with additional headers

Source code in katana_public_api_client/client.py 
46
47
48
49
50
51
52
def with_headers(self, headers: dict[str, str]) -> "Client":
    """Get a new client matching this one with additional headers"""
    if self._client is not None:
        self._client.headers.update(headers)
    if self._async_client is not None:
        self._async_client.headers.update(headers)
    return evolve(self, headers={**self._headers, **headers})
with_timeout(timeout)

Get a new client matching this one with a new timeout (in seconds)

Source code in katana_public_api_client/client.py 
62
63
64
65
66
67
68
def with_timeout(self, timeout: httpx.Timeout) -> "Client":
    """Get a new client matching this one with a new timeout (in seconds)"""
    if self._client is not None:
        self._client.timeout = timeout
    if self._async_client is not None:
        self._async_client.timeout = timeout
    return evolve(self, timeout=timeout)