roboto.http.response#

Module Contents#

class roboto.http.response.BatchResponse(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

The response to a batch request, holding one element per request element, in the order the request sent them.

Every element of a batch is applied or refused on its own, so a batch can come back partly applied. succeeded and failed split the outcomes for a caller that does not care which request element produced which; read responses to trace an outcome back to the request element at its position.

Parameters:

data (Any)

property failed: list[roboto.exceptions.RobotoDomainException]#

The exception the platform reported for each element it refused, in request order.

Return type:

list[roboto.exceptions.RobotoDomainException]

map_data(transform)#

Convert what each applied element carries, leaving positions and failures untouched.

A batch call parses the platform’s answer into records and uses this to hand the caller domain objects instead: a SessionRecord becomes a Session. transform runs only on elements carrying data; a refused element keeps its exception, and every element keeps its position.

Parameters:

transform (collections.abc.Callable[[Model], MappedModel]) – Builds the domain object an applied element’s record stands for.

Returns:

A batch holding one element per element of this one, in the same order.

Return type:

BatchResponse[MappedModel]

responses: list[BatchResponseElement[Model]]#
single()#

The result carried by the only element of a one-element batch.

A singular call such as create_session() sends its one element through the plural counterpart and unwraps the answer with this, so its caller gets a raised exception rather than a batch to inspect.

Raises:
Return type:

Model

property succeeded: list[Model]#

The result the platform returned for each element it applied, in request order.

Return type:

list[Model]

class roboto.http.response.BatchResponseElement(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

One element of a response to a batch request, holding data when the operation succeeded and error when it failed, never both. An element holding neither reports an operation that returns no content, the batch equivalent of a 204 answer to a singular call.

Parameters:

data (Any)

data: Model | None = None#
error: roboto.exceptions.RobotoDomainException | None = None#
model_config#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

serialize_error(value, info)#
Parameters:
Return type:

Optional[dict[str, Any]]

validate_error(value)#

Build the exception an element was refused with, in whichever form the failure arrived.

Three forms reach this: an exception object, which is what an element copied from another batch carries; the envelope RobotoDomainException.to_dict() produces; and that envelope as JSON text, which is what RobotoDomainException declares to pydantic. Only a literal None yields None, which is how an element the platform applied or answered with no content arrives.

Any other value is a refusal, whatever shape it has. An envelope naming an exception class this release does not define, one missing its code or message, JSON that is not an envelope, and text that is not JSON all read as a RobotoUnrecognizedErrorException carrying whatever code and message could be recovered, with the value’s own text as the message when no message could be. That way a refused element is always in BatchResponse.failed, so a caller checking failed before treating a batch as done cannot mistake a garbled refusal for success.

plain mode replaces the validation the error annotation would otherwise apply. That annotation reads JSON text, so it would reject the exception object returned here.

Parameters:

value (Any)

Return type:

Optional[roboto.exceptions.RobotoDomainException]

roboto.http.response.DEFAULT_RESPONSE_JSONPATH = ('data',)#
class roboto.http.response.HttpResponse(response)#
Parameters:

response (urllib.response.addinfourl)

property headers: dict[str, str] | None#
Return type:

Optional[dict[str, str]]

property readable_response: urllib.response.addinfourl#
Return type:

urllib.response.addinfourl

property status: http.HTTPStatus#
Return type:

http.HTTPStatus

to_dict(json_path=None)#
Parameters:

json_path (Optional[collections.abc.Sequence[str]])

Return type:

Any

to_int()#
Return type:

int

to_paginated_list(record_type)#
Parameters:

record_type (Type[PydanticModel])

Return type:

PaginatedList[PydanticModel]

to_record(record_type, json_path=DEFAULT_RESPONSE_JSONPATH)#
Parameters:

  • record_type (Type[PydanticModel])

  • json_path (Optional[collections.abc.Sequence[str]])

Return type:

PydanticModel

to_record_list(record_type, json_path=DEFAULT_RESPONSE_JSONPATH)#
Parameters:

  • record_type (Type[PydanticModel])

  • json_path (Optional[collections.abc.Sequence[str]])

Return type:

list[PydanticModel]

to_string()#
to_string_list()#
Return type:

list[str]

exception roboto.http.response.InvalidPaginationTokenError#

Bases: ValueError

Raised when a pagination token cannot be parsed.

A pagination token is opaque to clients, so one that fails to decode or carries an unsupported scheme reflects bad caller input — a fabricated, truncated, or stale token — rather than a server fault. Subclasses ValueError so existing callers that catch ValueError around PaginationToken.from_token() keep working unchanged, while callers that want to distinguish this recoverable input error (e.g. to surface an actionable message instead of a generic 500 or runtime exception) can catch it specifically.

roboto.http.response.MappedModel#
roboto.http.response.Model#
class roboto.http.response.PaginatedList(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

A list of records pulled from a paginated result set. It may be a subset of that result set, in which case next_token will be set and can be used to fetch the next page.

Parameters:

data (Any)

items: list[Model]#

Roboto entities in this page of results.

next_token: str | None = None#

Opaque token to fetch the next page of results.

If None, then this is the last page of results.

total_count: int | None = None#

Total result set size, if available.

class roboto.http.response.PaginationToken(scheme, encoding, data)#

A pagination token that can be treated as a truly opaque token by clients, with support for evolving the token format over time.

Parameters:
property data: Any#
Return type:

Any

static decode(data)#

Base64 decode the data, adding back any trailing padding (“=”) as necessary to make data properly Base64.

Parameters:

data (str)

Return type:

str

static empty()#
Return type:

PaginationToken

static encode(data)#

Base64 encode the data and strip all trailing padding (“=”).

Parameters:

data (str)

Return type:

str

classmethod from_token(token)#
Parameters:

token (Optional[str])

Return type:

PaginationToken

classmethod json_token(data)#
Parameters:

data (Any)

Return type:

PaginationToken

to_token()#
Return type:

str

class roboto.http.response.PaginationTokenEncoding(*args, **kwds)#

Bases: enum.Enum

Pagination token encoding enum

Json = 'json'#
Raw = 'raw'#
class roboto.http.response.PaginationTokenScheme(*args, **kwds)#

Bases: enum.Enum

Pagination token scheme enum

V1 = 'v1'#
roboto.http.response.PydanticModel#
class roboto.http.response.StreamedList(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

A StreamedList differs from a PaginatedList in that it represents a stream of data that is in process of being written to. Unlike a result set, which is finite and complete, a stream may be infinite, and it is unknown when or if it will complete.

Parameters:

data (Any)

has_next: bool#
items: list[Model]#
last_read: str | None#
roboto.http.response.logger#
On this page