swh.core.api.asynchronous module#

swh.core.api.asynchronous.encode_msgpack(data, **kwargs)[source]#
swh.core.api.asynchronous.render_msgpack(request, data, extra_encoders=None)[source]#
swh.core.api.asynchronous.render_json(request, data, extra_encoders=None)[source]#
swh.core.api.asynchronous.decode_data(data, content_type, extra_decoders=None)[source]#

Decode data according to content type, eventually using some extra decoders.

async swh.core.api.asynchronous.decode_request(request, extra_decoders=None)[source]#

Decode asynchronously the request

async swh.core.api.asynchronous.error_middleware(app, handler)[source]#
class swh.core.api.asynchronous.RPCServerApp(app_name: str | None = None, backend_class: Callable | None = None, backend_factory: Callable | str | None = None, middlewares=(), **kwargs)[source]#

Bases: Application

For each endpoint of the given backend_class, tells app.route to call a function that decodes the request and sends it to the backend object provided by the factory.

  • backend_class (Any) – The class of the backend, which will be analyzed to look for API endpoints.

  • backend_factory (Optional[Callable[[], backend_class]]) – A function with no argument that returns an instance of backend_class. If unset, defaults to calling backend_class constructor directly.

client_exception_classes: Tuple[Type[Exception], ...] = ()#

Exceptions that should be handled as a client error (eg. object not found, invalid argument)

extra_type_encoders: List[Tuple[type, str, Callable]] = []#

Value of extra_encoders passed to json_dumps or msgpack_dumps to be able to serialize more object types.

extra_type_decoders: Dict[str, Callable] = {}#

Value of extra_decoders passed to json_loads or msgpack_loads to be able to deserialize more object types.

class swh.core.api.asynchronous.SWHRemoteAPI(*args, **kwargs)[source]#

Bases: RPCServerApp