from __future__ import annotations
import random
from asyncio import sleep
from contextlib import asynccontextmanager
from ssl import SSLContext
from typing import Any, AsyncIterator, Callable, Literal, overload, override
from ._base import APIBaseEndpoint, APIClientBase
from ._errors import ErrorStatus, RetryError
from ._types import (
DEFAULT_LIMITS,
DEFAULT_MAX_REDIRECTS,
DEFAULT_TIMEOUT_CONFIG,
USE_CLIENT_DEFAULT,
AsyncBaseTransport,
AsyncClient,
AuthTypes,
CertTypes,
CookieTypes,
EventHook,
HeaderTypes,
HTTPMethods,
Limits,
Model,
ProxyTypes,
QueryParamTypes,
Request,
RequestContent,
RequestData,
RequestExtensions,
RequestFiles,
Response,
TimeoutTypes,
UseClientDefault,
XMLModel,
codes,
)
from ._utils import assign_annotations, unmarshal
class AsyncHTTPClientVerbs:
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model] | type[list[Model]] | None = None,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
headers: dict[str, str] | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
stream: bool = False,
) -> Model | list[Model] | Response:
raise NotImplementedError
@asynccontextmanager
async def _stream(
self,
method: HTTPMethods,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
headers: dict[str, str] | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> AsyncIterator[Response]:
"""
Construct and send an HTTP request and stream the response back. This is an
alternative method to making a call to _request that also supports closing of
the response through the use of a context manager.
Args:
method:
The HTTP method used to make the call.
path:
URL to query.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object with streaming enabled.
"""
response: Response = await self._request( # type: ignore[assignment] # ty: ignore[invalid-assignment]
method=method,
path=path,
params=params,
content=content,
data=data,
files=files,
headers=headers,
json=json,
xml=xml,
request_model_kwargs=request_model_kwargs,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
max_retries=max_retries,
error_map=error_map,
stream=True,
response_model=None,
response_model_kwargs=None,
)
try:
yield response
finally:
await response.aclose()
@overload
async def _get(
self,
path: str,
*,
response_model: None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
params: QueryParamTypes | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Response: ...
@overload
async def _get(
self,
path: str,
*,
response_model: type[Model],
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
params: QueryParamTypes | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Model: ...
@overload
async def _get(
self,
path: str,
*,
response_model: type[list[Model]],
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
params: QueryParamTypes | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> list[Model]: ...
async def _get(
self,
path: str,
*,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
params: QueryParamTypes | None = None,
headers: dict[str, str] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP GET request.
Args:
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
params:
Request query parameters.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
return await self._request( # ty: ignore[invalid-return-type]
method="GET",
path=path,
params=params,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
response_model=response_model,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
max_retries=max_retries,
error_map=error_map,
)
@overload
async def _post(
self,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Response: ...
@overload
async def _post(
self,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Model: ...
@overload
async def _post(
self,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> list[Model]: ...
async def _post(
self,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
headers: dict[str, str] | None = None,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP POST request.
Args:
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
return await self._request( # ty: ignore[invalid-return-type]
method="POST",
path=path,
params=params,
content=content,
data=data,
files=files,
json=json,
xml=xml,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
response_model=response_model,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
max_retries=max_retries,
error_map=error_map,
)
@overload
async def _put(
self,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Response: ...
@overload
async def _put(
self,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Model: ...
@overload
async def _put(
self,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> list[Model]: ...
async def _put(
self,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
headers: dict[str, str] | None = None,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP PUT request.
Args:
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
return await self._request( # ty: ignore[invalid-return-type]
method="PUT",
path=path,
params=params,
content=content,
data=data,
files=files,
json=json,
xml=xml,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
response_model=response_model,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
max_retries=max_retries,
error_map=error_map,
)
@overload
async def _patch(
self,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Response: ...
@overload
async def _patch(
self,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Model: ...
@overload
async def _patch(
self,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> list[Model]: ...
async def _patch(
self,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
headers: dict[str, str] | None = None,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP PATCH request.
Args:
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
return await self._request( # ty: ignore[invalid-return-type]
method="PATCH",
path=path,
params=params,
content=content,
data=data,
files=files,
json=json,
xml=xml,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
response_model=response_model,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
max_retries=max_retries,
error_map=error_map,
)
@overload
async def _delete(
self,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Response: ...
@overload
async def _delete(
self,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> Model: ...
@overload
async def _delete(
self,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
) -> list[Model]: ...
async def _delete(
self,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
headers: dict[str, str] | None = None,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP DELETE request.
Args:
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
return await self._request( # ty: ignore[invalid-return-type]
method="DELETE",
path=path,
params=params,
content=content,
data=data,
files=files,
json=json,
xml=xml,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
response_model=response_model,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
max_retries=max_retries,
error_map=error_map,
)
[docs]
class AsyncAPIEndpoint(APIBaseEndpoint, AsyncHTTPClientVerbs):
_client: AsyncAPIClient
def __init__(self, client: AsyncAPIClient | AsyncAPIEndpoint) -> None:
super().__init__(client)
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = ...,
) -> Response: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = False,
) -> Model: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = False,
) -> list[Model]: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model] | type[list[Model]] | None = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = ...,
) -> Model | list[Model] | Response: ...
[docs]
@override
async def _request(
self,
method: HTTPMethods,
path: str,
*,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
headers: dict[str, str] | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
response_model: type[Model] | type[list[Model]] | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
stream: bool = False,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP request.
Args:
method:
The HTTP method used to make the call.
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
if self._path is not None:
path = f"{self._path}{path}"
return await self._client._request( # ty: ignore[invalid-return-type]
method=method,
path=path,
params=params,
headers=headers,
cookies=cookies,
auth=auth,
follow_redirects=follow_redirects,
timeout=timeout,
extensions=extensions,
json=json,
xml=xml,
content=content,
data=data,
files=files,
response_model=response_model,
max_retries=max_retries,
response_model_kwargs=response_model_kwargs,
request_model_kwargs=request_model_kwargs,
error_map=error_map,
stream=stream,
)
[docs]
class AsyncAPIClient(APIClientBase, AsyncHTTPClientVerbs):
__client_class__: type[AsyncClient] = AsyncClient
__endpoint_class__: type[AsyncAPIEndpoint] = AsyncAPIEndpoint
_client: AsyncClient
@override
def __assign_annotations__(self) -> None:
"""
Any public annotations that are a subclass of the APIEndpoint class will be
assigned at initialization of the object through this method.
"""
assign_annotations(self, AsyncAPIEndpoint)
async def __aenter__(self):
"""
Async Context Manager __aenter__ dunder. See PEP-343 for more details.
"""
return self
async def __aexit__(self, exc_type, exc_value, exc_traceback):
"""
Async Context Manager __aexit__ dunder. See PEP-343 for more details.
"""
return self._deauthenticate()
def __init__(
self,
*,
auth: AuthTypes | None = None,
params: QueryParamTypes | None = None,
headers: HeaderTypes | None = None,
cookies: CookieTypes | None = None,
verify: SSLContext | str | bool = True,
cert: CertTypes | None = None,
http1: bool = True,
http2: bool = False,
proxy: ProxyTypes | None = None,
mounts: (dict[str, AsyncBaseTransport | None]) | None = None,
timeout: TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
limits: Limits = DEFAULT_LIMITS,
max_redirects: int = DEFAULT_MAX_REDIRECTS,
event_hooks: (dict[str, list[EventHook]]) | None = None,
base_url: str | None = None,
transport: AsyncBaseTransport | None = None,
trust_env: bool = True,
default_encoding: str | Callable[[bytes], str] = "utf-8",
vendor: str = "unknown",
product: str = "unknown",
build: str = "unknown",
retry_max: int = 5,
) -> None:
# Add the class event hooks to loop in the logging facilities.
event_hooks = {} if event_hooks is None else event_hooks
event_hooks = {
"request": [self._request_hook] + list(event_hooks.get("request", [])),
"response": [self._response_hook] + list(event_hooks.get("response", [])),
}
super().__init__(
auth=auth,
params=params,
headers=headers,
cookies=cookies,
verify=verify,
cert=cert,
http1=http1,
http2=http2,
proxy=proxy,
mounts=mounts,
timeout=timeout,
follow_redirects=follow_redirects,
limits=limits,
max_redirects=max_redirects,
event_hooks=event_hooks,
base_url=base_url,
transport=transport,
trust_env=trust_env,
default_encoding=default_encoding,
vendor=vendor,
product=product,
build=build,
retry_max=retry_max,
)
[docs]
async def _deauthenticate(self):
"""
De-authentication stub. De-authentication is automatically run as part
of leaving context within the context manager.
Example:
>>> class ExampleAPISession(APISession):
... def _deauthenticate(self):
... self.delete('session/token')
"""
[docs]
async def _request_hook(self, request: Request) -> None:
"""
The Request hook used for debug logging.
Args:
request: The request object.
"""
self._logger.debug(f"REQUESTING {request.method}: {request.url}")
[docs]
async def _response_hook(self, response: Response) -> None:
"""
The Response hook used for informational logging.
Args:
response: The response object.
"""
self._logger.info(
f"[{response.status_code}] {response.request.method}: {response.request.url}"
)
[docs]
async def _retry_request(self, response: Response) -> Request:
"""
Processes the request from the response for the purpose of retrying it again.
Args:
response: The response object
"""
return response.request
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: Literal[None] = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = ...,
) -> Response: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = False,
) -> Model: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[list[Model]],
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = False,
) -> list[Model]: ...
@overload
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model] | type[list[Model]] | None = ...,
params: QueryParamTypes | None = ...,
content: RequestContent | None = ...,
data: RequestData | None = ...,
files: RequestFiles | None = ...,
json: Model | Any | None = ...,
xml: XMLModel | str | bytes | None = ...,
response_model_kwargs: dict[str, Any] | None = ...,
request_model_kwargs: dict[str, Any] | None = ...,
headers: dict[str, str] | None = ...,
cookies: CookieTypes | None = ...,
auth: AuthTypes | UseClientDefault | None = ...,
follow_redirects: bool | UseClientDefault = ...,
timeout: TimeoutTypes | UseClientDefault = ...,
extensions: RequestExtensions | None = ...,
max_retries: int | None = ...,
error_map: dict[int, ErrorStatus] | None = ...,
stream: bool = ...,
) -> Model | list[Model] | Response: ...
[docs]
@override
async def _request(
self,
method: HTTPMethods,
path: str,
*,
response_model: type[Model] | type[list[Model]] | None = None,
params: QueryParamTypes | None = None,
content: RequestContent | None = None,
data: RequestData | None = None,
files: RequestFiles | None = None,
headers: dict[str, str] | None = None,
json: Model | Any | None = None,
xml: XMLModel | str | bytes | None = None,
response_model_kwargs: dict[str, Any] | None = None,
request_model_kwargs: dict[str, Any] | None = None,
cookies: CookieTypes | None = None,
auth: AuthTypes | UseClientDefault | None = USE_CLIENT_DEFAULT,
follow_redirects: bool | UseClientDefault = USE_CLIENT_DEFAULT,
timeout: TimeoutTypes | UseClientDefault = USE_CLIENT_DEFAULT,
extensions: RequestExtensions | None = None,
max_retries: int | None = None,
error_map: dict[int, ErrorStatus] | None = None,
stream: bool = False,
) -> Model | list[Model] | Response:
"""
Construct and send an HTTP request.
Args:
method:
The HTTP method used to make the call.
path:
URL to query.
response_model:
Pydantic model to coerce the response into.
response_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
un-marshalling the response data.
request_model_kwargs:
Keyword arguments to pass to Pydantic/Pydantic-XML as part of
marshalling the body of the data into the request.
params:
Request query parameters.
content:
Raw body of the request.
data:
URL form-encoded body for the request.
json:
Data object to marshal into JSON. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
xml:
Data object to marshal into XML. Content passed with this parameter
will also set the ``Content-Type`` header to ``application/json``.
files:
A dictionary of upload files to include in the body of the request.
headers:
Request-specific headers.
cookies:
Request-specific cookies.
auth:
Request-specific authentication.
follow_redirects:
Should the client follow any redirects?
timeout:
Request-specific timeout settings.
extensions:
Any additional httpx extensions to pass to the client.
max_retries:
The maximum number of retries to attempt before giving up. Overloads
the client default.
error_map:
Replaces the client error map with this one instead.
Returns:
Returns the HTTPX Response object if no response_model is specified. If a
response_model _is_ specified, then the response will be coerced into the
response model and the instance of the model will be returned.
"""
max_retries = max_retries if max_retries else self._retry_max
error_map = self._error_map if error_map is None else error_map
response_model_kwargs = (
{} if response_model_kwargs is None else response_model_kwargs
)
# Perform any pre-processing necessary on the request.
kwargs = self._request_pre_process(
method=method,
url=path,
params=params,
content=content,
data=data,
files=files,
json=json,
xml=xml,
headers=headers,
cookies=cookies,
timeout=timeout,
extensions=extensions,
request_model_kwargs=request_model_kwargs,
)
# Build the initial request and initialize the counter.
request = self._client.build_request(**kwargs)
request_counter = 0
# While the number of requests being performed is less than or equal to the
# maximum number allowed, then keep calling the API.
while request_counter <= max_retries:
request_counter += 1
response = await self._client.send(
request, auth=auth, follow_redirects=follow_redirects, stream=stream
)
# If the response is ok, then we should return the response. If a model is
# presented to us, then we will pass the model to the unmarshal utility to
# handle transitioning the data into the expected class.
if response.status_code == codes.OK and response_model:
return unmarshal(
response=response,
model=response_model,
json_model_kwargs=self._json_model_kwargs | response_model_kwargs,
xml_model_kwargs=self._xml_model_kwargs | response_model_kwargs,
)
# If no model was passed, then simply return the response object.
elif response.status_code == codes.OK:
return response
# As the response wasn't ok, let's grab the ErrorStatus object that relates
# to the status code that we got and determine the next steps.
status = error_map[response.status_code]
# If the status code is retryable, then pass the response to the retry
# handler to perform any optional transformation. Then sleep the amount
# if time determined by the status code and then continue to the next
# iteration.
if status.retry:
request = await self._retry_request(response)
timer = random.uniform(0, status.jitter) + (
request_counter * status.backoff
)
await sleep(timer)
continue
# Otherwise we will want to raise the error as specified by the error map
error_obj = None
raise status.exception(
response=response, template=status.template, obj=error_obj
)
# If too many attempts were made, then raise a retry error.
raise RetryError(url=str(path), method=method, attempts=request_counter)