slxjsonrpc
SlxJsonRpc is a JsonRpc helper class, that uses pydantic.
SlxJsonRpc keep track of the JsonRpc schema, and procedure for each method. It also ensures to route each message to where it is expected.
SlxJsonRpc is build to be both a JsonRpc server & client. To enable the JsonRpc-server, the method_map need to be given.
Installation using pip
The slxjsonrpc package can be installed using PIP (Python Package Index) as follows:
$ pip install slxjsonrpcUse case Examples
The given use case show how to use the slxJsonRpc Package. It expected that you have a send & a receive function/method to transport the Json RPC messages to and from the package.
The Client example code:
from typing import List, Union, Literal
from enum import Enum
import slxjsonrpc
def send(data: str) -> None: ...
def receive() -> Union[str, bytes, dict]: ...
class MethodList(str, Enum):
ADD = "add"
PING = "ping"
params = {
MethodList.ADD: List[Union[int, float]],
MethodList.PING: None,
}
result = {
MethodList.ADD: Union[int, float],
MethodList.PING: Literal["pong"]
}
client_jsonrpc = slxjsonrpc.SlxJsonRpc(
methods=MethodsList,
result=result,
params=params,
)
ok = None
def reply(reply_data):
nonlocal ok
ok = reply_data # Will be "pong"
ping_package = client_jsonrpc.create_request(method=MethodList.PING, callback=reply)
send(ping_package.json(exclude_none=True))
data = receive()
client_jsonrpc.parser(data)
print(f"OK: {ok}")The Server example code:
from typing import List, Union, Literal
from enum import Enum
import slxjsonrpc
def send(data: str) -> None: ...
def receive() -> Union[str, bytes, dict]: ...
class MethodList(str, Enum):
ADD = "add"
PING = "ping"
params = {
MethodList.ADD: List[Union[int, float]],
MethodList.PING: None,
}
result = {
MethodList.ADD: Union[int, float],
MethodList.PING: Literal["pong"]
}
method_map = {
MethodList.ADD: lambda data: sum(data),
MethodList.PING: lambda data: "pong",
}
server_jsonrpc = slxjsonrpc.SlxJsonRpc(
methods=MethodsList,
result=result,
params=params,
method_cb=method_map,
)
data = receive()
return_data = server_jsonrpc.parser(data)
if return_data:
send(return_data.json(exclude_none=True))License
This project is licensed under the Apache License 2.0 - see the LICENSE.md file for details.
