Package python_odpt
python-odpt
Python client for ODPT(Open Data of Public Transportation) API
Generated with OpenAPI spec on https://github.com/sophie-app/odpt-openapi using https://github.com/openapi-generators/openapi-python-client
Thanks for great works!
Installation
pip install python-odpt
Usage
WIP
A client library for accessing ODPT API
Sub-modules
python_odpt.api
-
Contains methods for accessing the API
python_odpt.client
python_odpt.errors
-
Contains shared errors types that can be raised from API functions
python_odpt.models
-
Contains all the data models used in inputs/outputs
python_odpt.types
-
Contains some shared types for properties
Classes
class AuthenticatedClient (base_url: str,
token: str,
prefix: str = 'Bearer',
auth_header_name: str = 'Authorization',
*,
raise_on_unexpected_status: bool = False,
cookies: Dict[str, str] = _Nothing.NOTHING,
headers: Dict[str, str] = _Nothing.NOTHING,
timeout: httpx.Timeout | None = None,
verify_ssl: str | bool | ssl.SSLContext = True,
follow_redirects: bool = False,
httpx_args: Dict[str, Any] = _Nothing.NOTHING)-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. token: The token to use for authentication prefix: The prefix to use for the Authorization header auth_header_name: The name of the Authorization header """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) token: str prefix: str = "Bearer" auth_header_name: str = "Authorization" 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
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:
<code>base\_url</code>: The base URL for the API, all requests are made to a relative path to this URL <code>cookies</code>: A dictionary of cookies to be sent with every request <code>headers</code>: A dictionary of headers to be sent with every request <code>timeout</code>: The maximum amount of a time a request can take. API functions will raise httpx.TimeoutException if this is exceeded. <code>verify\_ssl</code>: 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. <code>follow\_redirects</code>: Whether or not to follow redirects. Default value is False. <code>httpx\_args</code>: A dictionary of additional arguments to be passed to the <code>httpx.Client</code> and <code>httpx.AsyncClient</code> constructor.
Attributes
raise_on_unexpected_status
- Whether or not 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.
token
- The token to use for authentication
prefix
- The prefix to use for the Authorization header
auth_header_name
- The name of the Authorization header
Method generated by attrs for class AuthenticatedClient.
Instance variables
var auth_header_name : str
-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. token: The token to use for authentication prefix: The prefix to use for the Authorization header auth_header_name: The name of the Authorization header """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) token: str prefix: str = "Bearer" auth_header_name: str = "Authorization" 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
var prefix : str
-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. token: The token to use for authentication prefix: The prefix to use for the Authorization header auth_header_name: The name of the Authorization header """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) token: str prefix: str = "Bearer" auth_header_name: str = "Authorization" 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
var raise_on_unexpected_status : bool
-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. token: The token to use for authentication prefix: The prefix to use for the Authorization header auth_header_name: The name of the Authorization header """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) token: str prefix: str = "Bearer" auth_header_name: str = "Authorization" 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
var token : str
-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. token: The token to use for authentication prefix: The prefix to use for the Authorization header auth_header_name: The name of the Authorization header """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) token: str prefix: str = "Bearer" auth_header_name: str = "Authorization" 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
Methods
def get_async_httpx_client(self) ‑> httpx.AsyncClient
-
Expand source code
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 the underlying httpx.AsyncClient, constructing a new one if not previously set
def get_httpx_client(self) ‑> httpx.Client
-
Expand source code
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
Get the underlying httpx.Client, constructing a new one if not previously set
def set_async_httpx_client(self, async_client: httpx.AsyncClient) ‑> AuthenticatedClient
-
Expand source code
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
Manually the underlying httpx.AsyncClient
NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def set_httpx_client(self, client: httpx.Client) ‑> AuthenticatedClient
-
Expand source code
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
Manually set the underlying httpx.Client
NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
-
Expand source code
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})
Get a new client matching this one with additional cookies
def with_headers(self, headers: Dict[str, str]) ‑> AuthenticatedClient
-
Expand source code
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})
Get a new client matching this one with additional headers
def with_timeout(self, timeout: httpx.Timeout) ‑> AuthenticatedClient
-
Expand source code
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)
Get a new client matching this one with a new timeout (in seconds)
class Client (base_url: str,
*,
raise_on_unexpected_status: bool = False,
cookies: Dict[str, str] = _Nothing.NOTHING,
headers: Dict[str, str] = _Nothing.NOTHING,
timeout: httpx.Timeout | None = None,
verify_ssl: str | bool | ssl.SSLContext = True,
follow_redirects: bool = False,
httpx_args: Dict[str, Any] = _Nothing.NOTHING)-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
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:
<code>base\_url</code>: The base URL for the API, all requests are made to a relative path to this URL <code>cookies</code>: A dictionary of cookies to be sent with every request <code>headers</code>: A dictionary of headers to be sent with every request <code>timeout</code>: The maximum amount of a time a request can take. API functions will raise httpx.TimeoutException if this is exceeded. <code>verify\_ssl</code>: 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. <code>follow\_redirects</code>: Whether or not to follow redirects. Default value is False. <code>httpx\_args</code>: A dictionary of additional arguments to be passed to the <code>httpx.Client</code> and <code>httpx.AsyncClient</code> constructor.
Attributes
raise_on_unexpected_status
- Whether or not 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.
Method generated by attrs for class Client.
Instance variables
var raise_on_unexpected_status : bool
-
Expand source code
@define class 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. Attributes: raise_on_unexpected_status: Whether or not 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. """ raise_on_unexpected_status: bool = field(default=False, kw_only=True) _base_url: str = field(alias="base_url") _cookies: Dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") _headers: Dict[str, str] = field(factory=dict, kw_only=True, alias="headers") _timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") _verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") _follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") _httpx_args: Dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") _client: Optional[httpx.Client] = field(default=None, init=False) _async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) 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}) 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}) 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) 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 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 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 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) 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 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 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 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)
Methods
def get_async_httpx_client(self) ‑> httpx.AsyncClient
-
Expand source code
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 the underlying httpx.AsyncClient, constructing a new one if not previously set
def get_httpx_client(self) ‑> httpx.Client
-
Expand source code
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
Get the underlying httpx.Client, constructing a new one if not previously set
def set_async_httpx_client(self, async_client: httpx.AsyncClient) ‑> Client
-
Expand source code
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
Manually the underlying httpx.AsyncClient
NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
def set_httpx_client(self, client: httpx.Client) ‑> Client
-
Expand source code
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
Manually set the underlying httpx.Client
NOTE: This will override any other settings on the client, including cookies, headers, and timeout.
-
Expand source code
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})
Get a new client matching this one with additional cookies
def with_headers(self, headers: Dict[str, str]) ‑> Client
-
Expand source code
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})
Get a new client matching this one with additional headers
def with_timeout(self, timeout: httpx.Timeout) ‑> Client
-
Expand source code
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)
Get a new client matching this one with a new timeout (in seconds)