swh.core.api.asynchronous module

swh.core.api.asynchronous.encode_msgpack(data, **kwargs)

swh.core.api.asynchronous.render_msgpack(request, data, extra_encoders=None)

swh.core.api.asynchronous.render_json(request, data, extra_encoders=None)

swh.core.api.asynchronous.decode_data(data, content_type, extra_decoders=None)

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

async swh.core.api.asynchronous.decode_request(request, extra_decoders=None)

Decode asynchronously the request

async swh.core.api.asynchronous.error_middleware(app, handler)

class swh.core.api.asynchronous.RPCServerApp(app_name: str | None = None, backend_class: Callable | None = None, backend_factory: Callable | str | None = None, middlewares=(), **kwargs)

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.

Parameters:

• 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)

Bases: RPCServerApp