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. The responses element contains one response (either success data or failure error) per request element, in the order in which the request was sent.

Parameters:

data (Any)

responses: list[BatchResponseElement[Model]]#
class roboto.http.BatchResponseElement(/, **data)#

Bases: pydantic.BaseModel, Generic[Model]

One element of a response to a batch request. This should only ever have data set (in case of a successful operation) or error set (in case of a failed operation). For operations that do not return a response, an empty (data = None, error = None) Batch Response Element will be effectively equivalent to a single requests 204 No Content

Parameters:

data (Any)

data: Model | None = None#
error: roboto.exceptions.RobotoDomainException | None = None#
serialize_error(value, info)#
Parameters:
Return type:

Optional[dict[str, Any]]

validate_error(value)#
Parameters:

value (str)

Return type:

Optional[roboto.exceptions.RobotoDomainException]

class roboto.http.BearerTokenDecorator(token)#

Decorates requests with a static, unchanging bearer token.

Parameters:

token (str)

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)

class roboto.http.FakeHttpResponseFactory(url='https://iamverylazyanddonotseturls.com', response_data='{}', status_code=200, headers=None)#

A factory for creating fake HTTP responses, for use with the roboto.http.HttpClient.

Example

>>> import contextlib
>>> import unittest.mock
>>> from roboto.http import HttpClient, FakeHttpResponseFactory
>>> mock_http_client = unittest.mock.create_autospec(HttpClient, instance=True)
>>> with contextlib.ExitStack() as stack:
...     http_get_mock = stack.enter_context(
...         unittest.mock.patch.object(mock_http_client, "get")
...     )
...     http_get_mock.side_effect = FakeHttpResponseFactory(
...         "https://example.com",
...         {"foo": "bar"},
...         status_code=200,
...         headers={"Content-Type": "application/json"},
...     )
Parameters:

  • url (str)

  • response_data (Any)

  • status_code (int)

  • headers (Optional[dict[str, str]])

class roboto.http.HttpClient(base_headers=None, default_endpoint=None, default_auth=None, requester=None, extra_headers_provider=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]]])

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)#
Parameters:

  • url (str)

  • data (Any)

  • headers (Optional[dict])

  • idempotent (bool)

  • retry_wait (Optional[roboto.http.request.RetryWaitFn])

Return type:

roboto.http.response.HttpResponse

get(url, headers=None, retry_wait=None, idempotent=True)#
Parameters:

  • url (str)

  • headers (Optional[dict])

  • retry_wait (Optional[roboto.http.request.RetryWaitFn])

  • idempotent (bool)

Return type:

roboto.http.response.HttpResponse

patch(url, data=None, headers=None, idempotent=True, retry_wait=None)#
Parameters:

  • data (Any)

  • headers (Optional[dict])

  • idempotent (bool)

  • retry_wait (Optional[roboto.http.request.RetryWaitFn])

Return type:

roboto.http.response.HttpResponse

post(url, data=None, headers=None, idempotent=False, retry_wait=None)#
Parameters:

  • data (Any)

  • headers (Optional[dict])

  • idempotent (bool)

  • retry_wait (Optional[roboto.http.request.RetryWaitFn])

Return type:

roboto.http.response.HttpResponse

put(url, data=None, headers=None, idempotent=True, retry_wait=None)#
Parameters:

  • data (Any)

  • headers (Optional[dict])

  • idempotent (bool)

  • retry_wait (Optional[roboto.http.request.RetryWaitFn])

Return type:

roboto.http.response.HttpResponse

set_requester(requester)#
Parameters:

requester (roboto.http.requester.RobotoRequester)

url(path)#
Parameters:

path (str)

Return type:

str

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]

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]#
next_token: str | None = None#
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:
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#

Bases: enum.Enum

Pagination token encoding enum

Json = 'json'#
Raw = 'raw'#
class roboto.http.PaginationTokenScheme#

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.

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)#
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])

Return type:

roboto.http.response.HttpResponse

property endpoint: str#
Return type:

str

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)#
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])

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)#
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])

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)#
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])

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)#
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])

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: str, enum.Enum

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.roboto_headers(org_id=None, user_id=None, resource_owner_id=None, additional_headers=None)#
Parameters:

  • org_id (Optional[str])

  • user_id (Optional[str])

  • resource_owner_id (Optional[str])

  • additional_headers (Optional[dict[str, str]])

On this page