API Reference#
Important
This part of the documentation covers only public classes, their methods and attributes. Private objects are only needed for internal use and accessing of them by hand is discouraged.
ASGI Entry Point#
- class jsonrpc.ASGIHandler(dict=None, /, **kwargs)#
Base class representing the
ASGI
entry point. Its subclassing thecollections.UserDict
object for providing the user-defined data storage.For example:
>>> app = ASGIHandler() >>> app["my_private_key"] = "foobar" >>> app["my_private_key"] "foobar"
- content_type = 'application/json'#
The default content type of the responses.
- dispatcher = AsyncDispatcher()#
Class variable representing the
jsonrpc.AsyncDispatcher
object used by this class for routing user-defined functions by default.
- serializer = JSONSerializer()#
Class variable representing the
jsonrpc.JSONSerializer
object used by this class for data serialization by default.
- events = LifespanEvents(startup_events=set(), shutdown_events=set())#
Class variable representing the
jsonrpc.LifespanEvents
object used by this class for storing the user-defined functions that running when application is initiated and shutting down.
Routing user-defined functions#
- class jsonrpc.AsyncDispatcher(other=(), /, **kw)#
The
weakref.WeakValueDictionary
subclass representing the storage of user-defined functions.- register(user_function=None, *, function_name=None)#
Adds a user-defined function to the dispatcher.
Example usage:
>>> @dispatcher.register ... def truediv(a: float, b: float) -> float: ... return a / b
Also you can pass the different function’s name:
>>> @dispatcher.register(function_name="slow_greetings") ... async def _(name: str) -> str: ... return await sleep(3600.0, f"Greetings, {name} !!")
- Parameters:
user_function – The
types.FunctionType
object representing the user-defined function.function_name – An optional function’s name. If it is omitted, attribute
__name__
will be used instead.
- Raises:
RuntimeError – If the
user_function
isn’t passed by theinspect.isfunction()
method, or function with the provided name is already defined in thejsonrpc.AsyncDispatcher
class.- Returns:
The unmodified
user_function
object, passed in the parameters.
- async dispatch(function_name, /, *args, **kwargs)#
Invoke the user-defined function by passed in parameters function’s name.
Example usage:
>>> dispatcher = AsyncDispatcher() >>> await dispatcher.dispatch("sum", a=12, b=34) 46
- Parameters:
function_name – The user-defined function’s name.
args – Positional arguments for the provided function.
kwargs – Keyword arguments for the provided function.
- Raises:
jsonrpc.Error – If the function doesn’t exists in the
jsonrpc.AsyncDispatcher
class, passed invalid parameters or unexpected internal error has raised. See alsojsonrpc.ErrorEnum
.- Returns:
Result of execution the user-defined function.
Error handling#
- class jsonrpc.ErrorEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)#
An enumeration of error codes that indicates the error type that occurred.
- PARSE_ERROR = -32700#
Error occurred due the serialization or deserialization.
- INVALID_REQUEST = -32600#
Error occurred due the receiving an invalid
jsonrpc.Request
object.
- METHOD_NOT_FOUND = -32601#
Error occurred due the invoking a missing user-function.
- INVALID_PARAMETERS = -32602#
Error occurred due the receiving an invalid user-function’s arguments.
- INTERNAL_ERROR = -32603#
Error occurred due the unexpected internal errors.
- exception jsonrpc.Error(*, code, message, data=Undefined)#
Base class for all encountered errors in the JSON-RPC protocol.
- data#
An any type of object that contains additional information about the error.
- property json#
Returns the
dict
object needed for the serialization.Example output:
>>> error: Error = Error( ... code=ErrorEnum.INTERNAL_ERROR, ... message="Unexpected error", ... data={"additional": "information"} ... ) >>> error.json {"code": -32603, "message": "Unexpected error", "data": {"additional": "information"}}
Requests & Responses#
- class jsonrpc.Request(*, method, params=Undefined, request_id=Undefined)#
Base JSON-RPC request object.
- params#
The object of type
list
ordict
that holds the parameter values to be used during the invocation of the method. May be omitted if provided method has no parameters for example.
- request_id#
The
str
object or any type ofnumbers.Number
object which represents an identifier of the request instance. May be omitted. If its value omitted, the request assumed to be a notification.
- classmethod from_json(obj, /)#
The class method for creating the
jsonrpc.Request
object fromdict
object. Unlike thejsonrpc.Request
constructor, doesn’t raises any exceptions by validations, it returns thejsonrpc.Error
as is.Example usage:
>>> Request.from_json({"jsonrpc": "2.0", "method": "foobar", "id": 1}) Request(method="foobar", params=Undefined, request_id=1) >>> Request.from_json({"not_jsonrpc": True}) Error(code=-32600, message="Invalid request object", data={"not_jsonrpc": True})
- class jsonrpc.BatchRequest(initlist=None)#
The
collections.UserList
subclass representing the collection ofjsonrpc.Request
andjsonrpc.Error
objects.- classmethod from_json(iterable, /)#
The class method for creating the
jsonrpc.BatchRequest
object fromcollections.abc.Iterable
ofdict
objects. Similar tojsonrpc.Request.from_json()
function it doesn’t raises any exceptions.Example usage:
>>> BatchRequest.from_json([ ... {"jsonrpc": "2.0", "method": "foobar", "id": 1}, ... {"not_jsonrpc": True} ... ]) BatchRequest([Request(…), Error(…)])
- class jsonrpc.Response(*, body=Undefined, error=Undefined, response_id=Undefined)#
Base JSON-RPC response object.
- body#
An any type of object that contains a result of successful processing the
jsonrpc.Request
object. This attribute must not be set if there an error has occurred.
- error#
The
jsonrpc.Error
object representing an erroneous processing thejsonrpc.Request
object. This attribute must not be set if no one error has occurred.
- response_id#
The same attribute as
jsonrpc.Request.request_id
except that its value might be equal toNone
in erroneous responses.
- property json#
Returns the
dict
object needed for the serialization.Example successful response:
>>> response: Response = Response(body="foobar", response_id=65535) >>> response.json {"jsonrpc": "2.0", "result": "foobar", "id": 65535}
Example erroneous response:
>>> error: Error = Error(code=ErrorEnum.INTERNAL_ERROR, message="Unexpected error") >>> response: Response = Response(error=error, response_id="6ba7b810") >>> response.json {"jsonrpc": "2.0", "error": {"code": -32603, "message": "Unexpected error"}, "id": "6ba7b810"}
- class jsonrpc.BatchResponse(initlist=None)#
The
collections.UserList
subclass representing the collection ofjsonrpc.Response
objects.- property json#
Returns the
list
ofdict
objects needed for the serialization.Example output:
>>> response: BatchResponse = BatchResponse([ ... Response(body="foobar", response_id=1024), ... Response( ... error=Error(code=ErrorEnum.INTERNAL_ERROR, message="Unexpected error"), ... response_id="6ba7b810" ... ) ... ]) >>> response.json [ {"jsonrpc": "2.0", "result": "foobar", "id": 1024}, {"jsonrpc": "2.0", "error": {"code": -32603, "message": "Unexpected error"}, "id": "6ba7b810"} ]
Data Serialization#
- class jsonrpc.JSONSerializer#
Simple class for JSON serialization and deserialization.
- serialize(obj, /)#
Returns the JSON representation of a value.
- Parameters:
obj – An any type of object that must be JSON serializable.
- Raises:
jsonrpc.Error – If any exception has occurred due the serialization or/and encoding to
bytes
.- Returns:
The
bytes
object containing the serialized Python data structure.
- deserialize(obj, /)#
Returns the value encoded in JSON in appropriate Python type.
- Parameters:
obj – The
bytes
object containing the serialized JSON document.- Raises:
jsonrpc.Error – If any exception has occurred due the deserialization or/and decoding from
bytes
.- Returns:
An any type of object containing the deserialized Python data structure.
Lifespan#
- class jsonrpc.LifespanEvents(startup_events=<factory>, shutdown_events=<factory>)#
Simple class for storing the user-defined functions that running when application is initiated and shutting down.
- startup_events#
The
weakref.WeakSet
collection of startup functions.
- shutdown_events#
The
weakref.WeakSet
collection of shutdown functions.
- on_startup(user_function, /)#
Decorator for the adding a function which will be executed when application is initiated.
Example usage:
>>> @app.events.on_startup ... def startup_callback() -> None: ... print("Some important message.")
- Parameters:
user_function – The
collections.abc.Callable
object representing the user-defined function.- Returns:
The unmodified
user_function
object, passed in the parameters.
- on_shutdown(user_function, /)#
Decorator for the adding a function which will be executed when application is shutting down.
Example usage:
>>> @app.events.on_shutdown ... async def shutdown_callback() -> None: ... await important_function()
- Parameters:
user_function – The
collections.abc.Callable
object representing the user-defined function.- Returns:
The unmodified
user_function
object, passed in the parameters.