`nemo_microservices._response`#

Module Contents#

Classes#

`APIResponse`
`AsyncAPIResponse`
`AsyncBinaryAPIResponse`	Subclass of APIResponse providing helpers for dealing with binary data.
`AsyncResponseContextManager`	Context manager for ensuring that a request is not made until it is entered and that the response will always be closed when the context manager exits
`AsyncStreamedBinaryAPIResponse`
`BaseAPIResponse`
`BinaryAPIResponse`	Subclass of APIResponse providing helpers for dealing with binary data.
`ResponseContextManager`	Context manager for ensuring that a request is not made until it is entered and that the response will always be closed when the context manager exits
`StreamedBinaryAPIResponse`

Functions#

`async_to_custom_raw_response_wrapper`	Higher order function that takes one of our bound API methods and an `APIResponse` class and wraps the method to support returning the given response class directly.
`async_to_custom_streamed_response_wrapper`	Higher order function that takes one of our bound API methods and an `APIResponse` class and wraps the method to support streaming and returning the given response class directly.
`async_to_raw_response_wrapper`	Higher order function that takes one of our bound API methods and wraps it to support returning the raw `APIResponse` object directly.
`async_to_streamed_response_wrapper`	Higher order function that takes one of our bound API methods and wraps it to support streaming and returning the raw `APIResponse` object directly.
`extract_response_type`	Given a type like `APIResponse[T]`, returns the generic type variable `T`.
`to_custom_raw_response_wrapper`	Higher order function that takes one of our bound API methods and an `APIResponse` class and wraps the method to support returning the given response class directly.
`to_custom_streamed_response_wrapper`	Higher order function that takes one of our bound API methods and an `APIResponse` class and wraps the method to support streaming and returning the given response class directly.
`to_raw_response_wrapper`	Higher order function that takes one of our bound API methods and wraps it to support returning the raw `APIResponse` object directly.
`to_streamed_response_wrapper`	Higher order function that takes one of our bound API methods and wraps it to support streaming and returning the raw `APIResponse` object directly.

Data#

`P`
`R`
`log`

API#

class nemo_microservices._response.APIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.BaseAPIResponse[nemo_microservices._response.R]

close() → None#

Close the response and release the connection.

Automatically called if the response body is read to completion.

iter_bytes(chunk_size: int | None = None) → Iterator[bytes]#

A byte-iterator over the decoded response content.

This automatically handles gzip, deflate and brotli encoded responses.

iter_lines() → Iterator[str]#: Like iter_text() but will only yield chunks for each line

iter_text(chunk_size: int | None = None) → Iterator[str]#: A str-iterator over the decoded response content that handles both gzip, deflate, etc but also detects the content’s string encoding.

json() → object#: Read and decode the JSON response content.

parse( *, to: type[nemo_microservices._response._T] | None = None, ) → nemo_microservices._response.R | nemo_microservices._response._T#

Returns the rich python representation of this response’s data.

For lower-level control, see .read(), .json(), .iter_bytes().

You can customise the type that the response is parsed into through the to argument, e.g.

from nemo_microservices import BaseModel

class MyModel(BaseModel):
    foo: str

obj = response.parse(to=MyModel)
print(obj.foo)

We support parsing:

BaseModel
dict
list
Union
str
int
float
httpx.Response

read() → bytes#: Read and return the binary response content.

text() → str#: Read and decode the response content into a string.

class nemo_microservices._response.AsyncAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.BaseAPIResponse[nemo_microservices._response.R]

async close() → None#

Close the response and release the connection.

Automatically called if the response body is read to completion.

async iter_bytes( chunk_size: int | None = None, ) → AsyncIterator[bytes]#

A byte-iterator over the decoded response content.

This automatically handles gzip, deflate and brotli encoded responses.

async iter_lines() → AsyncIterator[str]#: Like iter_text() but will only yield chunks for each line

async iter_text(chunk_size: int | None = None) → AsyncIterator[str]#: A str-iterator over the decoded response content that handles both gzip, deflate, etc but also detects the content’s string encoding.

async json() → object#: Read and decode the JSON response content.

async parse( *, to: type[nemo_microservices._response._T] | None = None, ) → nemo_microservices._response.R | nemo_microservices._response._T#

Returns the rich python representation of this response’s data.

For lower-level control, see .read(), .json(), .iter_bytes().

You can customise the type that the response is parsed into through the to argument, e.g.

from nemo_microservices import BaseModel

class MyModel(BaseModel):
    foo: str

obj = response.parse(to=MyModel)
print(obj.foo)

We support parsing:

BaseModel
dict
list
Union
str
httpx.Response

async read() → bytes#: Read and return the binary response content.

async text() → str#: Read and decode the response content into a string.

class nemo_microservices._response.AsyncBinaryAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.AsyncAPIResponse[bytes]

Subclass of APIResponse providing helpers for dealing with binary data.

Note: If you want to stream the response data instead of eagerly reading it all at once then you should use .with_streaming_response when making the API request, e.g. .with_streaming_response.get_binary_response()

Initialization

async write_to_file(file: str | os.PathLike[str]) → None#

Write the output to the given file.

Accepts a filename or any path-like object, e.g. pathlib.Path

Note: if you want to stream the data to the file instead of writing all at once then you should use .with_streaming_response when making the API request, e.g. .with_streaming_response.get_binary_response()

class nemo_microservices._response.AsyncResponseContextManager( api_request: typing_extensions.Awaitable[nemo_microservices._response._AsyncAPIResponseT], )#

Bases: typing.Generic[nemo_microservices._response._AsyncAPIResponseT]

Context manager for ensuring that a request is not made until it is entered and that the response will always be closed when the context manager exits

Initialization

class nemo_microservices._response.AsyncStreamedBinaryAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.AsyncAPIResponse[bytes]

async stream_to_file( file: str | os.PathLike[str], *, chunk_size: int | None = None, ) → None#

Streams the output to the given file.

Accepts a filename or any path-like object, e.g. pathlib.Path

class nemo_microservices._response.BaseAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: typing.Generic[nemo_microservices._response.R]

property elapsed: datetime.timedelta#: The time taken for the complete request/response cycle to complete.

property headers: httpx.Headers#

property http_request: httpx.Request#: Returns the httpx Request instance associated with the current response.

http_response: httpx.Response#: None

property http_version: str#

property is_closed: bool#

Whether or not the response body has been closed.

If this is False then there is response data that has not been read yet. You must either fully consume the response body or call .close() before discarding the response to prevent resource leaks.

property method: str#

retries_taken: int#

None

The number of retries made. If no retries happened this will be 0

property status_code: int#

property url: httpx.URL#: Returns the URL for which the request was made.

class nemo_microservices._response.BinaryAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.APIResponse[bytes]

Subclass of APIResponse providing helpers for dealing with binary data.

Note: If you want to stream the response data instead of eagerly reading it all at once then you should use .with_streaming_response when making the API request, e.g. .with_streaming_response.get_binary_response()

Initialization

write_to_file(file: str | os.PathLike[str]) → None#

Write the output to the given file.

Accepts a filename or any path-like object, e.g. pathlib.Path

Note: if you want to stream the data to the file instead of writing all at once then you should use .with_streaming_response when making the API request, e.g. .with_streaming_response.get_binary_response()

exception nemo_microservices._response.MissingStreamClassError#: Bases: TypeError

nemo_microservices._response.P#: ‘ParamSpec(…)’

nemo_microservices._response.R#: ‘TypeVar(…)’

class nemo_microservices._response.ResponseContextManager( request_func: Callable[[], nemo_microservices._response._APIResponseT], )#

Bases: typing.Generic[nemo_microservices._response._APIResponseT]

Context manager for ensuring that a request is not made until it is entered and that the response will always be closed when the context manager exits

Initialization

exception nemo_microservices._response.StreamAlreadyConsumed#

Bases: nemo_microservices._exceptions.NeMoMicroservicesError

Attempted to read or stream content, but the content has already been streamed.

This can happen if you use a method like .iter_lines() and then attempt to read th entire response body afterwards, e.g.

response = await client.post(...)
async for line in response.iter_lines():
    ...  # do something with `line`

content = await response.read()
# ^ error

If you want this behaviour you’ll need to either manually accumulate the response content or call await response.read() before iterating over the stream.

Initialization

Initialize self. See help(type(self)) for accurate signature.

class nemo_microservices._response.StreamedBinaryAPIResponse( *, raw: httpx.Response, cast_to: type[nemo_microservices._response.R], client: nemo_microservices._base_client.BaseClient[Any, Any], stream: bool, stream_cls: type[nemo_microservices._streaming.Stream[Any]] | type[nemo_microservices._streaming.AsyncStream[Any]] | None, options: nemo_microservices._models.FinalRequestOptions, retries_taken: int = 0, )#

Bases: nemo_microservices._response.APIResponse[bytes]

stream_to_file( file: str | os.PathLike[str], *, chunk_size: int | None = None, ) → None#

Streams the output to the given file.

Accepts a filename or any path-like object, e.g. pathlib.Path

nemo_microservices._response.async_to_custom_raw_response_wrapper( func: Callable[nemo_microservices._response.P, typing_extensions.Awaitable[object]], response_cls: type[nemo_microservices._response._AsyncAPIResponseT], ) → Callable[nemo_microservices._response.P, typing_extensions.Awaitable[nemo_microservices._response._AsyncAPIResponseT]]#

Higher order function that takes one of our bound API methods and an APIResponse class and wraps the method to support returning the given response class directly.

Note: the given response_cls must be concrete, e.g. class BinaryAPIResponse(APIResponse[bytes])

nemo_microservices._response.async_to_custom_streamed_response_wrapper( func: Callable[nemo_microservices._response.P, typing_extensions.Awaitable[object]], response_cls: type[nemo_microservices._response._AsyncAPIResponseT], ) → Callable[nemo_microservices._response.P, nemo_microservices._response.AsyncResponseContextManager[nemo_microservices._response._AsyncAPIResponseT]]#

Higher order function that takes one of our bound API methods and an APIResponse class and wraps the method to support streaming and returning the given response class directly.

Note: the given response_cls must be concrete, e.g. class BinaryAPIResponse(APIResponse[bytes])

nemo_microservices._response.async_to_raw_response_wrapper( func: Callable[nemo_microservices._response.P, typing_extensions.Awaitable[nemo_microservices._response.R]], ) → Callable[nemo_microservices._response.P, typing_extensions.Awaitable[nemo_microservices._response.AsyncAPIResponse[nemo_microservices._response.R]]]#: Higher order function that takes one of our bound API methods and wraps it to support returning the raw APIResponse object directly.

nemo_microservices._response.async_to_streamed_response_wrapper( func: Callable[nemo_microservices._response.P, typing_extensions.Awaitable[nemo_microservices._response.R]], ) → Callable[nemo_microservices._response.P, nemo_microservices._response.AsyncResponseContextManager[nemo_microservices._response.AsyncAPIResponse[nemo_microservices._response.R]]]#: Higher order function that takes one of our bound API methods and wraps it to support streaming and returning the raw APIResponse object directly.

nemo_microservices._response.extract_response_type( typ: type[nemo_microservices._response.BaseAPIResponse[Any]], ) → type#

Given a type like APIResponse[T], returns the generic type variable T.

This also handles the case where a concrete subclass is given, e.g.

class MyResponse(APIResponse[bytes]):
    ...

extract_response_type(MyResponse) -> bytes

nemo_microservices._response.log: logging.Logger#: ‘getLogger(…)’

nemo_microservices._response.to_custom_raw_response_wrapper( func: Callable[nemo_microservices._response.P, object], response_cls: type[nemo_microservices._response._APIResponseT], ) → Callable[nemo_microservices._response.P, nemo_microservices._response._APIResponseT]#

Higher order function that takes one of our bound API methods and an APIResponse class and wraps the method to support returning the given response class directly.

Note: the given response_cls must be concrete, e.g. class BinaryAPIResponse(APIResponse[bytes])

nemo_microservices._response.to_custom_streamed_response_wrapper( func: Callable[nemo_microservices._response.P, object], response_cls: type[nemo_microservices._response._APIResponseT], ) → Callable[nemo_microservices._response.P, nemo_microservices._response.ResponseContextManager[nemo_microservices._response._APIResponseT]]#

Higher order function that takes one of our bound API methods and an APIResponse class and wraps the method to support streaming and returning the given response class directly.

Note: the given response_cls must be concrete, e.g. class BinaryAPIResponse(APIResponse[bytes])

nemo_microservices._response.to_raw_response_wrapper( func: Callable[nemo_microservices._response.P, nemo_microservices._response.R], ) → Callable[nemo_microservices._response.P, nemo_microservices._response.APIResponse[nemo_microservices._response.R]]#: Higher order function that takes one of our bound API methods and wraps it to support returning the raw APIResponse object directly.

nemo_microservices._response.to_streamed_response_wrapper( func: Callable[nemo_microservices._response.P, nemo_microservices._response.R], ) → Callable[nemo_microservices._response.P, nemo_microservices._response.ResponseContextManager[nemo_microservices._response.APIResponse[nemo_microservices._response.R]]]#: Higher order function that takes one of our bound API methods and wraps it to support streaming and returning the raw APIResponse object directly.

nemo_microservices._response#

Module Contents#

Classes#

Functions#

Data#

API#

`nemo_microservices._response`#