pdstools.infinity.internal._base_client¶

Attributes¶

`_HttpxClientT`
`logger`
`ResponseT`
`DefaultAsyncHttpxClient`	An alias to httpx.AsyncClient that provides the same defaults that this SDK

Classes¶

`BaseClient`	Abstract base class for generic types.
`SyncAPIClient`	Abstract base class for generic types.
`_DefaultAsyncHttpxClient`	An asynchronous HTTP client, with connection pooling, HTTP/2, redirects,
`AsyncHttpxClientWrapper`	An asynchronous HTTP client, with connection pooling, HTTP/2, redirects,
`AsyncAPIClient`	Abstract base class for generic types.

Functions¶

`execute_and_collect`(task_coro, results, i)
`get_results`(→ List[Any])

Module Contents¶

_HttpxClientT¶

logger¶

ResponseT¶

async execute_and_collect(task_coro: Coroutine, results: List, i: int)¶

Parameters:

task_coro (Coroutine)
results (List)
i (int)

async get_results(tasks: List[Coroutine]) → List[Any]¶

Parameters:: tasks (List[Coroutine])
Return type:: List[Any]

class BaseClient(*, base_url: str | httpx.URL, auth: httpx.Auth | pdstools.infinity.internal._auth.PegaOAuth, verify: bool = False, pega_version: str | None = None, timeout: float = 20)¶

Bases: Generic[_HttpxClientT]

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as:

class Mapping(Generic[KT, VT]):
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

This class can then be used as follows:

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Parameters:

base_url (Union[str, httpx.URL])
auth (Union[httpx.Auth, pdstools.infinity.internal._auth.PegaOAuth])
verify (bool)
pega_version (Union[str, None])
timeout (float)

_client: _HttpxClientT¶

_base_url¶

auth¶

verify = False¶

pega_version = None¶

timeout = 20¶

_enforce_trailing_slash(url: httpx.URL) → httpx.URL¶

Parameters:: url (httpx.URL)
Return type:: httpx.URL

_build_request(method, endpoint: str, data: httpx._types.RequestData | None = None, **params) → httpx.Request¶

Parameters:

endpoint (str)
data (Union[httpx._types.RequestData, None])

Return type:

httpx.Request

_get_version(repo)¶

classmethod from_client_credentials(file_path: str, verify: bool = False, pega_version: str | None = None, timeout: float = 20)¶

Parameters:

file_path (str)
verify (bool)
pega_version (Union[str, None])
timeout (float)

classmethod from_basic_auth(base_url: str | None = None, user_name: str | None = None, password: str | None = None, *, verify: bool = True, pega_version: str | None = None, timeout: int = 20)¶

Parameters:

base_url (Optional[str])
user_name (Optional[str])
password (Optional[str])
verify (bool)
pega_version (Union[str, None])
timeout (int)

class SyncAPIClient(base_url: str | httpx.URL, auth: httpx.Auth | pdstools.infinity.internal._auth.PegaOAuth, verify: bool = False, pega_version: str | None = None, timeout: float = 20)¶

Bases: BaseClient[httpx.Client]

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as:

class Mapping(Generic[KT, VT]):
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

This class can then be used as follows:

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Parameters:

base_url (Union[str, httpx.URL])
auth (Union[httpx.Auth, pdstools.infinity.internal._auth.PegaOAuth])
verify (bool)
pega_version (Union[str, None])
timeout (float)

_client: httpx.Client¶

_infer_version()¶

_request(*, method, endpoint, data: httpx._types.RequestData | None = None, **params) → httpx.Response¶

Parameters:: data (Union[httpx._types.RequestData, None])
Return type:: httpx.Response

handle_pega_exception(endpoint, params, response)¶

request(method, endpoint, **params)¶

get(endpoint: str, **params)¶

Parameters:: endpoint (str)

post(endpoint: str, data: httpx._types.RequestData | None = None, **params)¶

Parameters:

endpoint (str)
data (Union[httpx._types.RequestData, None])

patch(endpoint, data: httpx._types.RequestData | None = None, **params)¶

Parameters:: data (Union[httpx._types.RequestData, None])

put(endpoint, data: httpx._types.RequestData | None = None, **params)¶

Parameters:: data (Union[httpx._types.RequestData, None])

abstract delete()¶

abstract get_api_list()¶

class _DefaultAsyncHttpxClient(**kwargs: Any)¶

Bases: httpx.AsyncClient

An asynchronous HTTP client, with connection pooling, HTTP/2, redirects, cookie persistence, etc.

It can be shared between tasks.

Usage:

`python >>> async with httpx.AsyncClient() as client: >>> response = await client.get('https://example.org') `

Parameters:

auth - (optional) An authentication class to use when sending

requests. * params - (optional) Query parameters to include in request URLs, as a string, dictionary, or sequence of two-tuples. * headers - (optional) Dictionary of HTTP headers to include when sending requests. * cookies - (optional) Dictionary of Cookie items to include when sending requests. * verify - (optional) Either True to use an SSL context with the default CA bundle, False to disable verification, or an instance of ssl.SSLContext to use a custom context. * http2 - (optional) A boolean indicating if HTTP/2 support should be enabled. Defaults to False. * proxy - (optional) A proxy URL where all the traffic should be routed. * timeout - (optional) The timeout configuration to use when sending requests. * limits - (optional) The limits configuration to use. * max_redirects - (optional) The maximum number of redirect responses that should be followed. * base_url - (optional) A URL to use as the base when building request URLs. * transport - (optional) A transport class to use for sending requests over the network. * trust_env - (optional) Enables or disables usage of environment variables for configuration. * default_encoding - (optional) The default encoding to use for decoding response text, if no charset information is included in a response Content-Type header. Set to a callable for automatic character set detection. Default: “utf-8”.

Parameters:: kwargs (Any)

DefaultAsyncHttpxClient¶

An alias to httpx.AsyncClient that provides the same defaults that this SDK uses internally.

This is useful because overriding the http_client with your own instance of httpx.AsyncClient will result in httpx’s defaults being used, not ours.

class AsyncHttpxClientWrapper(*, auth: httpx._types.AuthTypes | None = None, params: httpx._types.QueryParamTypes | None = None, headers: httpx._types.HeaderTypes | None = None, cookies: httpx._types.CookieTypes | None = None, verify: ssl.SSLContext | str | bool = True, cert: httpx._types.CertTypes | None = None, http1: bool = True, http2: bool = False, proxy: httpx._types.ProxyTypes | None = None, mounts: None | Mapping[str, httpx._transports.base.AsyncBaseTransport | None] = None, timeout: httpx._types.TimeoutTypes = DEFAULT_TIMEOUT_CONFIG, follow_redirects: bool = False, limits: httpx._config.Limits = DEFAULT_LIMITS, max_redirects: int = DEFAULT_MAX_REDIRECTS, event_hooks: None | Mapping[str, list[EventHook]] = None, base_url: httpx._urls.URL | str = '', transport: httpx._transports.base.AsyncBaseTransport | None = None, trust_env: bool = True, default_encoding: str | Callable[[bytes], str] = 'utf-8')¶

Bases: DefaultAsyncHttpxClient

An asynchronous HTTP client, with connection pooling, HTTP/2, redirects, cookie persistence, etc.

It can be shared between tasks.

Usage:

`python >>> async with httpx.AsyncClient() as client: >>> response = await client.get('https://example.org') `

Parameters:

auth - (optional) An authentication class to use when sending

Parameters:

auth (httpx._types.AuthTypes | None)
params (httpx._types.QueryParamTypes | None)
headers (httpx._types.HeaderTypes | None)
cookies (httpx._types.CookieTypes | None)
verify (ssl.SSLContext | str | bool)
cert (httpx._types.CertTypes | None)
http1 (bool)
http2 (bool)
proxy (httpx._types.ProxyTypes | None)
mounts (None | Mapping[str, httpx._transports.base.AsyncBaseTransport | None])
timeout (httpx._types.TimeoutTypes)
follow_redirects (bool)
limits (httpx._config.Limits)
max_redirects (int)
event_hooks (None | Mapping[str, list[EventHook]])
base_url (httpx._urls.URL | str)
transport (httpx._transports.base.AsyncBaseTransport | None)
trust_env (bool)
default_encoding (str | Callable[[bytes], str])

__del__() → None¶

Return type:: None

class AsyncAPIClient(base_url: str | httpx.URL, auth: httpx.Auth | pdstools.infinity.internal._auth.PegaOAuth, verify: bool = False, pega_version: str | None = None)¶

Bases: BaseClient[httpx.AsyncClient]

Abstract base class for generic types.

A generic type is typically declared by inheriting from this class parameterized with one or more type variables. For example, a generic mapping type might be defined as:

class Mapping(Generic[KT, VT]):
    def __getitem__(self, key: KT) -> VT:
        ...
    # Etc.

This class can then be used as follows:

def lookup_name(mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
    try:
        return mapping[key]
    except KeyError:
        return default

Parameters:

base_url (Union[str, httpx.URL])
auth (Union[httpx.Auth, pdstools.infinity.internal._auth.PegaOAuth])
verify (bool)
pega_version (Union[str, None])

_client: httpx.AsyncClient¶

_collect_awaitable_blocking(coros: List[Coroutine] | Coroutine) → Any¶

Parameters:: coros (Union[List[Coroutine], Coroutine])
Return type:: Any

_infer_version()¶

async _request(*, method, endpoint, **params) → httpx.Response¶

Return type:: httpx.Response

async request(method, endpoint, **params)¶

async get(endpoint: str, **params)¶

Parameters:: endpoint (str)

abstract post()¶

abstract patch()¶

abstract put()¶

abstract delete()¶

abstract get_api_list()¶