roboto.http#

Submodules#

Package Contents#

roboto.http.BEARER_TOKEN_HEADER = 'X-Roboto-Bearer-Token'#: Bearer token which is parsed as a JWT to provide additional request context for invocations

class roboto.http.BatchRequest(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

Batched HTTP requests

Parameters:: data (Any)

requests: list[Model]#

class roboto.http.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:

RobotoDomainException – Whatever the platform refused the element with.
RobotoInternalException – The batch does not hold exactly one element, or holds one carrying neither a result nor an error.

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.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:

value (Optional[roboto.exceptions.RobotoDomainException])
info (pydantic.SerializationInfo)

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]

class roboto.http.BearerTokenDecorator(token)#

Decorates requests with a static, unchanging bearer token.

Parameters:: token (str)

roboto.http.CONNECTION_CONSISTENCY_HEADER = 'X-Roboto-Connection-Consistency'#

Header to control database read consistency for query/search endpoints.

Set to ‘strongly_consistent’ to force strongly consistent reads, or ‘eventually_consistent’ to force eventually consistent reads. Omission defaults to the API version’s default.

Only applies to query and search endpoints (e.g., /v1/query/*, /v1/datasets/query, /v1/files/query). Has no effect on other endpoints.

For API versions >= 2026-03-13, query endpoints default to eventually consistent reads. For older API versions, they default to strongly consistent reads for backward compatibility.

roboto.http.CONTENT_TYPE_JSON_HEADER#

exception roboto.http.ClientError(exc)#

Bases: HttpError

Common base class for all non-exit exceptions.

Parameters:: exc (urllib.error.HTTPError)

roboto.http.DEFAULT_HTTP_TIMEOUT = 30.0#

class roboto.http.HttpClient(base_headers=None, default_endpoint=None, default_auth=None, requester=None, extra_headers_provider=None, default_timeout=None, options=None)#

Parameters:

base_headers (Optional[dict[str, str]])
default_endpoint (Optional[str])
default_auth (Optional[roboto.http.request.HttpRequestDecorator])
requester (Optional[roboto.http.requester.RobotoRequester])
extra_headers_provider (Optional[Callable[[], dict[str, str]]])
default_timeout (Optional[float])
options (Optional[roboto.http.options.HttpClientOptions])

property auth_decorator: roboto.http.request.HttpRequestDecorator | None#

Return type:: Optional[roboto.http.request.HttpRequestDecorator]

delete(url, data=None, headers=None, idempotent=True, retry_wait=None, timeout=NotSet)#

Parameters:

url (str)
data (Any)
headers (Optional[dict])
idempotent (bool)
retry_wait (Optional[roboto.http.request.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

get(url, headers=None, retry_wait=None, idempotent=True, timeout=NotSet)#

Parameters:

url (str)
headers (Optional[dict])
retry_wait (Optional[roboto.http.request.RetryWaitFn])
idempotent (bool)
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

patch(url, data=None, headers=None, idempotent=True, retry_wait=None, timeout=NotSet)#

Parameters:

url (str)
data (Any)
headers (Optional[dict])
idempotent (bool)
retry_wait (Optional[roboto.http.request.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

post(url, data=None, headers=None, idempotent=False, retry_wait=None, timeout=NotSet)#

Parameters:

url (str)
data (Any)
headers (Optional[dict])
idempotent (bool)
retry_wait (Optional[roboto.http.request.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

put(url, data=None, headers=None, idempotent=True, retry_wait=None, timeout=NotSet)#

Parameters:

url (str)
data (Any)
headers (Optional[dict])
idempotent (bool)
retry_wait (Optional[roboto.http.request.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

set_requester(requester)#

Parameters:: requester (roboto.http.requester.RobotoRequester)

url(path)#

Parameters:: path (str)
Return type:: str

class roboto.http.HttpClientOptions#

Behavior of a HttpClient, applied to every request it makes.

logging: HttpLoggingOptions#

retry: HttpRetryOptions#

exception roboto.http.HttpError(exc)#

Bases: Exception

Common base class for all non-exit exceptions.

Parameters:: exc (urllib.error.HTTPError)

property headers: dict#

Return type:: dict

property msg: Any#

Return type:: Any

property status: http.HTTPStatus | None#

Return type:: Optional[http.HTTPStatus]

class roboto.http.HttpLoggingOptions#

How a HttpClient renders requests into its logs.

scrub_headers: Sequence[str] = ()#: Header names, compared case-insensitively, whose values are replaced with * in every rendered form of a request the client produces. Authorization is always scrubbed, with or without an entry here.

class roboto.http.HttpRequest(url, method='GET', headers=None, data=None, retry_wait=None, idempotent=False)#

Parameters:

url (str)
method (str)
headers (Optional[dict[str, str]])
data (Any)
retry_wait (Optional[roboto.http.retry.RetryWaitFn])
idempotent (bool)

append_headers(headers)#

Parameters:: headers (dict[str, str])
Return type:: None

property body: bytes | None#

Return type:: Optional[bytes]

data: Any = None#

describe(scrub_headers=())#

Render the request for logging, with sensitive header values replaced by *.

Parameters:: scrub_headers (Collection[str]) – Header names, compared case-insensitively, to scrub in addition to ALWAYS_SCRUBBED_HEADERS.
Return type:: str

headers: dict#

property hostname: str#

Return type:: str

idempotent: bool = False#

method: str#

retry_wait: roboto.http.retry.RetryWaitFn#

url: str#

class roboto.http.HttpRetryOptions#

Whether and how many times a HttpClient retries a failed request.

max_attempts: int = 10#

the first attempt always runs.

Type:: Total attempts per request, first try included. Values below 1 behave as 1

predicate: RetryPredicate | None = None#: Called with the request and the exception a failed attempt raised; True means try again. None keeps the default, is_expected_to_be_transient(), which retries failures expected to be transient: DNS resolution failures unconditionally; connection errors, timeouts, and retryable HTTP statuses according to the request’s idempotency.

exception roboto.http.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.ORG_OVERRIDE_HEADER = 'X-Roboto-Org-Id'#: Header to specify the organization that the user is acting on behalf of.

roboto.http.ORG_OVERRIDE_QUERY_PARAM = 'robotoOrgId'#: Query parameter to specify the organization that the user is acting on behalf of.

class roboto.http.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.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:

scheme (PaginationTokenScheme)
encoding (PaginationTokenEncoding)
data (Any)

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.PaginationTokenEncoding(*args, **kwds)#

Bases: enum.Enum

Pagination token encoding enum

Json = 'json'#

Raw = 'raw'#

class roboto.http.PaginationTokenScheme(*args, **kwds)#

Bases: enum.Enum

Pagination token scheme enum

V1 = 'v1'#

roboto.http.RESOURCE_OWNER_OVERRIDE_HEADER = 'X-Roboto-Resource-Owner-Id'#: Header to specify the organization that owns the resource being accessed.

roboto.http.RESOURCE_OWNER_OVERRIDE_QUERY_PARAM = 'robotoResourceOwnerId'#: Query parameter to specify the organization that owns the resource being accessed.

roboto.http.ROBOTO_REQUESTER_HEADER = 'X-Roboto-Requester'#: A JSON serialized RobotoRequester representing the entity making a request to Roboto.

roboto.http.RetryPredicate#: Decides whether the exception warrants another attempt at the request.

class roboto.http.RobotoClient(endpoint, auth_decorator, http_client_kwargs=None)#

A client for making HTTP requests against Roboto service

Parameters:

endpoint (str)
auth_decorator (Optional[roboto.http.request.HttpRequestDecorator])
http_client_kwargs (Optional[dict[str, Any]])

classmethod defaulted(client=None)#

Parameters:: client (Optional[RobotoClient])
Return type:: RobotoClient

delete(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=NotSet)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

property endpoint: str#

Return type:: str

classmethod for_profile(profile)#

Parameters:: profile (str)
Return type:: RobotoClient

classmethod from_config(config)#

Parameters:: config (roboto.config.RobotoConfig)
Return type:: RobotoClient

classmethod from_env()#

Return type:: RobotoClient

property frontend_endpoint: str#

Return type:: str

get(path, caller_org_id=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=NotSet)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

property http_client: roboto.http.http_client.HttpClient#

Return type:: roboto.http.http_client.HttpClient

patch(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=NotSet)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

post(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=NotSet)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

put(path, caller_org_id=None, data=None, headers=None, idempotent=True, owner_org_id=None, query=None, retry_wait_fn=None, timeout=NotSet)#

Parameters:

path (ApiRelativePath)
caller_org_id (Optional[str])
data (Any)
headers (Optional[dict[str, str]])
idempotent (bool)
owner_org_id (Optional[str])
query (Optional[dict[str, Any]])
retry_wait_fn (Optional[roboto.http.retry.RetryWaitFn])
timeout (roboto.env.Timeout)

Return type:

roboto.http.response.HttpResponse

class roboto.http.RobotoRequester(/, **data)#

Bases: pydantic.BaseModel

Details about the entity making a request to Roboto. These are embedded in a header in order to see what tool versions / operating systems are making requests, and to aid debugging.

Parameters:: data (Any)

classmethod for_tool(tool)#

Called to intelligently populate a RobotoRequester for a request made from a named Roboto tool using the Python SDK.

Parameters:: tool (RobotoTool)
Return type:: RobotoRequester

platform: str | None = None#: The environment in which a request is being made, i.e. the user agent (for browser requests) or the results of platform.platform (for SDK requests)

roboto_tool: RobotoTool | str | None = None#: If a request is being made from a Roboto vended tool, the name of the tool

roboto_tool_details: str | None = None#: If a request is being made from a Roboto vended tool, free text pertinent details about the tool

roboto_tool_version: str | None = None#: If a request is being made from a Roboto vended tool, the version of the tool

schema_version: Literal['v1'] = 'v1'#: Roboto Requester payload schema version, used to ensure backward compatibility

class roboto.http.RobotoTool#

Bases: roboto.compat.StrEnum

Tool used to access Roboto

Cli = 'cli'#

Sdk = 'sdk'#

UploadAgent = 'upload-agent'#

Website = 'website'#

exception roboto.http.ServerError(exc)#

Bases: HttpError

Common base class for all non-exit exceptions.

Parameters:: exc (urllib.error.HTTPError)

class roboto.http.SigV4AuthDecorator(service='execute-api', credentials=None, region=None)#

Parameters:

service (str)
credentials (Optional[botocore.credentials.ReadOnlyCredentials])
region (Optional[str])

static lookup_credentials()#

Return type:: botocore.credentials.ReadOnlyCredentials

static lookup_region()#

Return type:: str

class roboto.http.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.USER_OVERRIDE_HEADER = 'X-Roboto-User-Id'#: Header to specify the user that is performing the REST operation.

roboto.http.USER_OVERRIDE_QUERY_PARAM = 'robotoUserId'#: “Query parameter to specify the user that is performing the REST operation.

roboto.http.is_expected_to_be_transient(request, exc)#

The default retry predicate: whether exc is expected to be transient for request.

Parameters:

request (roboto.http.request.HttpRequest)
exc (BaseException)

Return type:

bool

roboto.http.never_retry(_request, _exc)#

Retry predicate giving every request exactly one attempt.

Use for calls whose side effect must not run twice and where the server offers no idempotency key: a retried delivery (a chat message, an email) lands as a duplicate.

Parameters:

_request (roboto.http.request.HttpRequest)
_exc (BaseException)

Return type:

bool

roboto.http.roboto_headers(org_id=None, user_id=None, resource_owner_id=None, additional_headers=None, api_version=RobotoApiVersion.latest())#

Parameters:

org_id (Optional[str])
user_id (Optional[str])
resource_owner_id (Optional[str])
additional_headers (Optional[dict[str, str]])
api_version (roboto.api_version.RobotoApiVersion)