wxc_sdk.as_rest module

REST session for Webex API requests

class wxc_sdk.as_rest.AsErrorMessage(*, description: str, code: int | None = None, errorCode: int | None = None, **extra_data: Any)[source]

Bases: ApiModel

description: str
code: int | None
error_code: int | None
class wxc_sdk.as_rest.AsSingleError(*, key: str | None = None, message: list[AsErrorMessage], **extra_data: Any)[source]

Bases: ApiModel

Representation of single error in the body of an HTTP error response from Webex

key: str | None
message: list[AsErrorMessage]
property code: int | None

Error code or None

Returns:

error code

property description: str | None

Description or None

Returns:

description

class wxc_sdk.as_rest.AsErrorDetail(*, error: list[AsSingleError] | None = None, errorCode: int | None = None, trackingId: str | None = None, message: str | None = None, errors: list[AsErrorMessage] | None = None, **extra_data: Any)[source]

Bases: ApiModel

Representation of error details in the body of an HTTP error response from Webex. There are several variants of error responses. This model tries to generalize them

error: list[AsSingleError] | None
error_code: int | None
tracking_id: str | None
message: str | None
errors: list[AsErrorMessage] | None
property description: str

error description

property code: int | None

error code

exception wxc_sdk.as_rest.AsRestError(request_info: RequestInfo, history: tuple[ClientResponse, ...], *, code: int | None = None, status: int | None = None, message: str = '', headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, detail: Any = None)[source]

Bases: ClientResponseError

A REST error

__init__(request_info: RequestInfo, history: tuple[ClientResponse, ...], *, code: int | None = None, status: int | None = None, message: str = '', headers: Mapping[str, str] | Mapping[istr, str] | CIMultiDict | CIMultiDictProxy | Iterable[Tuple[str | istr, str]] | None = None, detail: Any = None) → None[source]
wxc_sdk.as_rest.as_dump_response(*, response: ClientResponse, response_data: Any = None, request_body: str = None, file: TextIOBase = None, dump_log: Logger = None, diff_ns: int = None) → None[source]

Dump response object to log file

Parameters:

  • response – HTTP request response

  • response_data

  • request_body

  • file (TextIOBase) – stream to dump to

  • dump_log (logging.Logger) – logger to dump to

  • diff_ns (int) – time the request took (in ns)

class wxc_sdk.as_rest.AsRestSession(*, tokens: Tokens, concurrent_requests: int, retry_429: bool = True, trace_configs: list[TraceConfig] = None, proxy_url: str = None, ssl: bool | Fingerprint | SSLContext = None, **kwargs: Any)[source]

Bases: ClientSession

REST session used for API requests:

  • includes an Authorization header in reach request

  • implements retries on 429

  • loads deserializes JSON data if needed

BASE = 'https://webexapis.com/v1'

base URL for all Webex API requests

__init__(*, tokens: Tokens, concurrent_requests: int, retry_429: bool = True, trace_configs: list[TraceConfig] = None, proxy_url: str = None, ssl: bool | Fingerprint | SSLContext = None, **kwargs: Any)[source]

Initialize the REST session

Parameters:

  • tokens – tokens to be used for the session

  • concurrent_requests – maximum number of concurrent requests

  • retry_429 – enable automatic retry on 429 responses

  • trace_configs – trace configurations, passed to aiohttp.ClientSession

  • proxy_url – used as proxy argument for all aiohttp.ClientSession.request() calls

  • ssl – used as ssl argument for all aiohttp.ClientSession.request() calls

  • kwargs – additional arguments. All arguments with a req_ prefix are passed to each :meth:`aiohttp.ClientSession.request call. All other arguments are passed to the constructor of aiohttp.ClientSession

retry_429: bool
register_response_callback(callback: Callable[[ClientResponse, str, str, dict[Any, Any], int], None]) → str[source]

Register a response callback

Registered response callbacks are called with the response object and the time the request took for all responses. This can be used for logging or other purposes.

Parameters:

callback – callback to register

Returns:

callback ID

unregister_response_callback(id: str) → None[source]

Unregister a response callback

Parameters:

id – callback ID

property access_token: str | None

access token used for all requests

Returns:

access token

Return type:

str

async rest_get(*args: Any, **kwargs: Any) → str | dict[str, Any][source]

GET request

Parameters:

  • args

  • kwargs

Returns:

deserialized JSON content or body text

async rest_post(*args: Any, **kwargs: Any) → str | dict[str, Any][source]

POST request

Parameters:

  • args

  • kwargs

Returns:

deserialized JSON content or body text

async rest_put(*args: Any, **kwargs: Any) → str | dict[str, Any][source]

PUT request

Parameters:

  • args

  • kwargs

Returns:

deserialized JSON content or body text

async rest_delete(*args: Any, **kwargs: Any) → str | dict[str, Any][source]

DELETE request

Parameters:

  • args

  • kwargs

async rest_patch(*args: Any, **kwargs: Any) → str | dict[str, Any][source]

PATCH request

Parameters:

  • args

  • kwargs

async follow_pagination(url: str, model: type[ApiModelType] = None, params: dict[str, Any] = None, item_key: str = None, **kwargs: Any) → AsyncGenerator[ApiModelType, None][source]

Handling RFC5988 pagination of list requests. Generator of parsed objects

Parameters:

  • url (str) – start url for 1st GET

  • model (ApiModel) – data type to return

  • params (Optional[dict]) – URL parameters, optional

  • item_key (str) – key to list of values

Returns:

yields parsed objects