pykern.api package

HTTP WebSocket Quest server & client

copyright:

Copyright (c) 2024 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

Submodules

pykern.api.auth_api module

?

copyright:

Copyright (c) 2025 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

class pykern.api.auth_api.AuthAPI(*args, **kwargs)

Bases: API

VERSION = 658584001

Defaults version number but allows override

async api_authenticate_connection(api_args)

Process AuthRequest from server

api_args:

token (str): secret value evaluated by AuthAPI version (int): protocol version

Parameters:

api_args (PKDict) – what to validate

Returns:

validation result

Return type:

Result

token()

pykern.api.client module

WebSocket Quest client

copyright:

Copyright (c) 2025 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

class pykern.api.client.Client(http_config)

Bases: object

Wrapper for tornado.httpclient.AsyncHTTPClient

Maybe called as an async context manager

http_config.request_config is deprecated.

Parameters:

http_config (PKDict) – tcp_ip, tcp_port, api_uri

async call_api(api_name, api_args)

Make a request to the API server

Parameters:

• api_name (str) – what to call on the server

• api_args (PKDict) – passed verbatim to the API on the server.

Returns:

value of api_result.

Return type:

PKDict

Raises:

• pykern.util.APIError – if there was an raise in the API or on a server protocol violation

• Exception – other exceptions that AsyncHTTPClient.fetch may raise, e.g. NotFound

async connect(auth_args=None)

Connect to the server

Parameters:

auth_args (PKDict) – how to authenticate connection; may be AuthArgs or other PKDict [None]

Returns:

self

Return type:

Client

destroy()

Must be called

remove_call(call_id)

Not a public interface

async subscribe_api(api_name, api_args)

Subscribe to api_name from API server

Maybe used in with:

with client.subscribe_api(api, args) as s:
    while (r := await s.result_get()):
        process r

Alternately, you must call _Call.unsubscribe:

s = client.subscribe_api(api, args)
r = await s.result_get()
... process r and possibly more calls to result_get ...
s.unsubscribe()

Parameters:

• api_name (str) – what to call on the server

• api_args (PKDict) – passed verbatim to the API on the server.

Returns:

to get replies or unsubscribe

Return type:

_Call

unsubscribe_call(call_id)

Not a public interface

pykern.api.server module

WebSocket Quest server

copyright:

Copyright (c) 2025 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

class pykern.api.server.Session(qcall, **kwargs)

Bases: Attr

State held on server bound to a client.

Currently the state is not persisted when the server terminates. This may change.

ATTR_KEY = 'session'

IS_SINGLETON = True

shared Attrs do not have link to qcall

handle_on_close()

class pykern.api.server.Subscription(server_msg)

Bases: Attr

EXPERIMENTAL

ATTR_KEY = 'subscription'

result_put(api_result)

pykern.api.server.start(api_classes, attr_classes, http_config, coros=())

Start _Server in pkasyncio

Parameters:

• api_classes (Iterable) – pykern.quest.API subclasses to be dispatched

• attr_classes (Iterable) – pykern.quest.Attr subclasses to create API instance

• http_config (PKDict) – pkasyncio.Loop.http_server arg

• coros (Iterable) – list of coroutines to be passed to pkasyncio.Loop.run

pykern.api.unit_util module

Support for pykern.api tests

copyright:

Copyright (c) 2024 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

class pykern.api.unit_util.Setup(**server_config)

Bases: object

Usage:

async with http_unit.Setup(api_classes=(_class())) as c:
    from pykern.pkcollections import PKDict
    from pykern import pkunit

    e = PKDict(ping="pong")
    pkunit.pkeq(e.pkupdate(counter=1), await c.call_api("echo", e))
    pkunit.pkeq(e.pkupdate(counter=2), await c.call_api("echo", e))

May be subclassed to start multiple servers.

destroy()

Destroy client and kill attributes *_pid attrs

pykern.api.util module

API constants

copyright:

Copyright (c) 2025 RadiaSoft LLC. All Rights Reserved.

license:

http://www.apache.org/licenses/LICENSE-2.0.html

exception pykern.api.util.APICallError(error)

Bases: APIError

Raised when call execution ends in exception or other error

exception pykern.api.util.APIDisconnected

Bases: APIError

Raised when remote server closed or other error

exception pykern.api.util.APIForbidden

Bases: APIError

Raised for forbidden or protocol error

exception pykern.api.util.APIKindError(error)

Bases: APIError

Raised when kind mismatch

exception pykern.api.util.APINotFound(api_name)

Bases: APIError

Raised for an object not found

exception pykern.api.util.APIProtocolError(error)

Bases: APIError

Raised when protocol error at lower level

pykern.api.util.AUTH_API_NAME = 'authenticate_connection'

API that authenticates connections (needed for client)

pykern.api.util.AUTH_API_VERSION = 658584001

API version for AUTH (and for pykern.api)

class pykern.api.util.MsgKind(*values)

Bases: Enum

CALL = 777501

REPLY = 777502

SUBSCRIBE = 777503

UNSUBSCRIBE = 777504

is_call()

is_reply()

is_subscribe()

is_unsubscribe()

pykern.api.util.is_subscription(func)

Is func a subscription api?

Parameters:

func (function) – class api

Returns:

True if is subscription api

Return type:

bool

pykern.api.util.msg_pack(unserialized)

Used by client and server, not public

pykern.api.util.msg_unpack(serialized, which)

Used by client and server, not public

pykern.api.util.subscription(func)

Decorator for api functions thhat can be subscribed by clients.

Parameters:

func (function) – class api

Returns:

function to use

Return type:

function