nemo_microservices._base_client#

Module Contents#

Classes#

AsyncAPIClient

AsyncHttpxClientWrapper

AsyncPaginator

BaseAsyncPage

BaseClient

BasePage

Defines the core interface for pagination.

BaseSyncPage

ForceMultipartDict

OtherArch

OtherPlatform

PageInfo

Stores the necessary information to build the request to retrieve the next page.

SyncAPIClient

SyncHttpxClientWrapper

Functions#

get_architecture

get_platform

get_python_runtime

get_python_version

make_request_options

Create a dict of type RequestOptions without keys of NotGiven values.

platform_headers

Data#

API#

nemo_microservices._base_client.Arch#

None

class nemo_microservices._base_client.AsyncAPIClient(
*,
version: str,
base_url: str | httpx.URL,
_strict_response_validation: bool,
max_retries: int = DEFAULT_MAX_RETRIES,
timeout: float | nemo_microservices._types.Timeout | None | nemo_microservices._types.NotGiven = not_given,
http_client: httpx.AsyncClient | None = None,
custom_headers: Mapping[str, str] | None = None,
custom_query: Mapping[str, object] | None = None,
)#

Bases: nemo_microservices._base_client.BaseClient[httpx.AsyncClient, nemo_microservices._streaming.AsyncStream[typing.Any]]

async close() None#

Close the underlying HTTPX client.

The client will not be usable after this.

async delete(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
async get(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
options: nemo_microservices._types.RequestOptions = {},
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._AsyncStreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._AsyncStreamT#
get_api_list(
path: str,
*,
model: Type[nemo_microservices._base_client._T],
page: Type[nemo_microservices._base_client.AsyncPageT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
method: str = 'get',
) nemo_microservices._base_client.AsyncPaginator[nemo_microservices._base_client._T, nemo_microservices._base_client.AsyncPageT]#
is_closed() bool#
async patch(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
async post(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
files: nemo_microservices._types.RequestFiles | None = None,
options: nemo_microservices._types.RequestOptions = {},
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._AsyncStreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._AsyncStreamT#
async put(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
files: nemo_microservices._types.RequestFiles | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
async request(
cast_to: Type[nemo_microservices._types.ResponseT],
options: nemo_microservices._models.FinalRequestOptions,
*,
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._AsyncStreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._AsyncStreamT#
class nemo_microservices._base_client.AsyncHttpxClientWrapper(
*,
auth: httpx._types.AuthTypes | None = None,
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
verify: httpx._types.VerifyTypes = True,
cert: httpx._types.CertTypes | None = None,
http1: bool = True,
http2: bool = False,
proxy: httpx._types.ProxyTypes | None = None,
proxies: httpx._types.ProxiesTypes | None = None,
mounts: None | Mapping[str, httpx._transports.base.AsyncBaseTransport | None] = None,
timeout: httpx._types.TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
limits: httpx._config.Limits = DEFAULT_LIMITS,
max_redirects: int = DEFAULT_MAX_REDIRECTS,
event_hooks: None | Mapping[str, list[httpx._client.EventHook]] = None,
base_url: httpx._urls.URL | str = '',
transport: httpx._transports.base.AsyncBaseTransport | None = None,
app: Callable[..., Any] | None = None,
trust_env: bool = True,
default_encoding: str | Callable[[bytes], str] = 'utf-8',
)#

Bases: nemo_microservices._base_client.DefaultAsyncHttpxClient

nemo_microservices._base_client.AsyncPageT#

‘TypeVar(…)’

class nemo_microservices._base_client.AsyncPaginator(
client: nemo_microservices._base_client.AsyncAPIClient,
options: nemo_microservices._models.FinalRequestOptions,
page_cls: Type[nemo_microservices._base_client.AsyncPageT],
model: Type[nemo_microservices._base_client._T],
)#

Bases: typing.Generic[nemo_microservices._base_client._T, nemo_microservices._base_client.AsyncPageT]

class nemo_microservices._base_client.BaseAsyncPage#

Bases: nemo_microservices._base_client.BasePage[nemo_microservices._base_client._T], typing.Generic[nemo_microservices._base_client._T]

async get_next_page() nemo_microservices._base_client.AsyncPageT#
async iter_pages() AsyncIterator[nemo_microservices._base_client.AsyncPageT]#
class nemo_microservices._base_client.BaseClient(
*,
version: str,
base_url: str | httpx.URL,
_strict_response_validation: bool,
max_retries: int = DEFAULT_MAX_RETRIES,
timeout: float | nemo_microservices._types.Timeout | None = DEFAULT_TIMEOUT,
custom_headers: Mapping[str, str] | None = None,
custom_query: Mapping[str, object] | None = None,
)#

Bases: typing.Generic[nemo_microservices._base_client._HttpxClientT, nemo_microservices._base_client._DefaultStreamT]

property auth_headers: dict[str, str]#
property base_url: httpx.URL#
property custom_auth: httpx.Auth | None#
property default_headers: dict[str, str | nemo_microservices._types.Omit]#
property default_query: dict[str, object]#
max_retries: int#

None

platform_headers() Dict[str, str]#
property qs: nemo_microservices._qs.Querystring#
timeout: float | nemo_microservices._types.Timeout | None#

None

property user_agent: str#
class nemo_microservices._base_client.BasePage(/, **data: Any)#

Bases: nemo_microservices._models.GenericModel, typing.Generic[nemo_microservices._base_client._T]

Defines the core interface for pagination.

Type Args: ModelT: The pydantic model that represents an item in the response.

Methods: has_next_page(): Check if there is another page available next_page_info(): Get the necessary information to make a request for the next page

Initialization

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

has_next_page() bool#
next_page_info() nemo_microservices._base_client.PageInfo | None#
class nemo_microservices._base_client.BaseSyncPage#

Bases: nemo_microservices._base_client.BasePage[nemo_microservices._base_client._T], typing.Generic[nemo_microservices._base_client._T]

get_next_page() nemo_microservices._base_client.SyncPageT#
iter_pages() Iterator[nemo_microservices._base_client.SyncPageT]#
class nemo_microservices._base_client.ForceMultipartDict#

Bases: typing.Dict[str, None]

class nemo_microservices._base_client.OtherArch(name: str)#

Initialization

class nemo_microservices._base_client.OtherPlatform(name: str)#

Initialization

class nemo_microservices._base_client.PageInfo(
*,
url: httpx.URL | nemo_microservices._types.NotGiven = not_given,
json: nemo_microservices._types.Body | nemo_microservices._types.NotGiven = not_given,
params: nemo_microservices._types.Query | nemo_microservices._types.NotGiven = not_given,
)#

Stores the necessary information to build the request to retrieve the next page.

Either url or params must be set.

Initialization

json: nemo_microservices._types.Body | nemo_microservices._types.NotGiven#

None

params: nemo_microservices._types.Query | nemo_microservices._types.NotGiven#

None

url: httpx.URL | nemo_microservices._types.NotGiven#

None

nemo_microservices._base_client.Platform#

None

class nemo_microservices._base_client.SyncAPIClient(
*,
version: str,
base_url: str | httpx.URL,
max_retries: int = DEFAULT_MAX_RETRIES,
timeout: float | nemo_microservices._types.Timeout | None | nemo_microservices._types.NotGiven = not_given,
http_client: httpx.Client | None = None,
custom_headers: Mapping[str, str] | None = None,
custom_query: Mapping[str, object] | None = None,
_strict_response_validation: bool,
)#

Bases: nemo_microservices._base_client.BaseClient[httpx.Client, nemo_microservices._streaming.Stream[typing.Any]]

close() None#

Close the underlying HTTPX client.

The client will not be usable after this.

delete(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
get(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
options: nemo_microservices._types.RequestOptions = {},
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._StreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._StreamT#
get_api_list(
path: str,
*,
model: Type[object],
page: Type[nemo_microservices._base_client.SyncPageT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
method: str = 'get',
) nemo_microservices._base_client.SyncPageT#
is_closed() bool#
patch(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
post(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
options: nemo_microservices._types.RequestOptions = {},
files: nemo_microservices._types.RequestFiles | None = None,
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._StreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._StreamT#
put(
path: str,
*,
cast_to: Type[nemo_microservices._types.ResponseT],
body: nemo_microservices._types.Body | None = None,
files: nemo_microservices._types.RequestFiles | None = None,
options: nemo_microservices._types.RequestOptions = {},
) nemo_microservices._types.ResponseT#
request(
cast_to: Type[nemo_microservices._types.ResponseT],
options: nemo_microservices._models.FinalRequestOptions,
*,
stream: bool = False,
stream_cls: type[nemo_microservices._base_client._StreamT] | None = None,
) nemo_microservices._types.ResponseT | nemo_microservices._base_client._StreamT#
class nemo_microservices._base_client.SyncHttpxClientWrapper(
*,
auth: httpx._types.AuthTypes | None = None,
params: httpx._types.QueryParamTypes | None = None,
headers: httpx._types.HeaderTypes | None = None,
cookies: httpx._types.CookieTypes | None = None,
verify: httpx._types.VerifyTypes = True,
cert: httpx._types.CertTypes | None = None,
http1: bool = True,
http2: bool = False,
proxy: httpx._types.ProxyTypes | None = None,
proxies: httpx._types.ProxiesTypes | None = None,
mounts: None | Mapping[str, httpx._transports.base.BaseTransport | None] = None,
timeout: httpx._types.TimeoutTypes = DEFAULT_TIMEOUT_CONFIG,
follow_redirects: bool = False,
limits: httpx._config.Limits = DEFAULT_LIMITS,
max_redirects: int = DEFAULT_MAX_REDIRECTS,
event_hooks: None | Mapping[str, list[httpx._client.EventHook]] = None,
base_url: httpx._urls.URL | str = '',
transport: httpx._transports.base.BaseTransport | None = None,
app: Callable[..., Any] | None = None,
trust_env: bool = True,
default_encoding: str | Callable[[bytes], str] = 'utf-8',
)#

Bases: nemo_microservices._base_client.DefaultHttpxClient

nemo_microservices._base_client.SyncPageT#

‘TypeVar(…)’

nemo_microservices._base_client.get_architecture() nemo_microservices._base_client.Arch#
nemo_microservices._base_client.get_platform() nemo_microservices._base_client.Platform#
nemo_microservices._base_client.get_python_runtime() str#
nemo_microservices._base_client.get_python_version() str#
nemo_microservices._base_client.log: logging.Logger#

‘getLogger(…)’

nemo_microservices._base_client.make_request_options(
*,
query: nemo_microservices._types.Query | None = None,
extra_headers: nemo_microservices._types.Headers | None = None,
extra_query: nemo_microservices._types.Query | None = None,
extra_body: nemo_microservices._types.Body | None = None,
idempotency_key: str | None = None,
timeout: float | httpx.Timeout | None | nemo_microservices._types.NotGiven = not_given,
post_parser: nemo_microservices._types.PostParser | nemo_microservices._types.NotGiven = not_given,
) nemo_microservices._types.RequestOptions#

Create a dict of type RequestOptions without keys of NotGiven values.

nemo_microservices._base_client.platform_headers(
version: str,
*,
platform: nemo_microservices._base_client.Platform | None,
) Dict[str, str]#