jsonrpyc

jsonrpyc#

Minimal python RPC implementation based on the JSON-RPC 2.0 specs.

Original source hosted at GitHub.

Usage#

jsonrpyc.RPC instances basically wrap an input stream and an output stream in order to communicate with other services. A service is not even forced to be written in Python as long as it strictly implements the JSON-RPC 2.0 specs. A suitable implementation for NodeJs is node-json-rpc. A jsonrpyc.RPC instance may wrap a target object. Incomming requests will be routed to methods of this object whose result might be sent back as a response. Example implementation:

`server.py`#

import jsonrpyc

class MyTarget(object):

    def greet(self: MyTarget, name: str) -> str:
        return f"Hi, {name}!"

jsonrpyc.RPC(MyTarget())

`client.py`#

import jsonrpyc
from subprocess import Popen, PIPE

p = Popen(["python", "server.py"], stdin=PIPE, stdout=PIPE)
rpc = jsonrpyc.RPC(stdout=p.stdin, stdin=p.stdout)


#
# sync usage
#

print(rpc("greet", args=("John",), block=0.1))
# => "Hi, John!"


#
# async usage
#

def cb(err: Exception | None, res: str | None = None) -> None:
    if err:
        raise err
    print(f"callback got: {res}")

rpc("greet", args=("John",), callback=cb)

# cb is called asynchronously which prints
# => "callback got: Hi, John!"


#
# shutdown
#

p.stdin.close()
p.stdout.close()
p.terminate()
p.wait()

API Docs#

Minimal python RPC implementation in a single file based on the JSON-RPC 2.0 specs from http://www.jsonrpc.org/specification.

Classes#

class RPC(target: Any | None = None, stdin: InputStream | None = None, stdout: OutputStream | None = None, *, watch: bool = True, watch_kwargs: dict[str, Any] | None = None)[source]#

The main class of jsonrpyc. Instances of this class wrap an input stream stdin and an output stream stdout in order to communicate with other services. A service is not even forced to be written in Python as long as it strictly implements the JSON-RPC 2.0 specification. RPC instances may wrap a target object. By means of a Watchdog instance, incoming requests are routed to methods of this object whose result might be sent back as a response. The watchdog instance is created but not started yet, when watch is not True.

Parameters:

target – The target object to wrap.
stdin – The input stream.
stdout – The output stream.
watch – Whether to start the watchdog.
watch_kwargs – Additional keyword arguments for the watchdog.

Example implementation:

server.py

import jsonrpyc

class MyTarget(object):

    def greet(self, name):
        return f"Hi, {name}!"

jsonrpc.RPC(MyTarget())

client.py

import jsonrpyc
from subprocess import Popen, PIPE

p = Popen(["python", "server.py"], stdin=PIPE, stdout=PIPE)
rpc = jsonrpyc.RPC(stdout=p.stdin, stdin=p.stdout)

# non-blocking remote procedure call with callback and js-like signature
def cb(err, res=None):
    if err:
        throw err
    print(f"callback got: {res}")

rpc("greet", args=("John",), callback=cb)

# cb is called asynchronously which prints
# => "callback got: Hi, John!"

# blocking remote procedure call with 0.1s polling
print(rpc("greet", args=("John",), block=0.1))
# => "Hi, John!"

# shutdown the process
p.stdin.close()
p.stdout.close()
p.terminate()
p.wait()

target#: The wrapped target object. Might be None when no object is wrapped, e.g. for the client RPC instance.

stdin#: The input stream, re-opened with "rb".

stdout#: The output stream, re-opened with "wb".

watch#: The Watchdog instance that optionally watches stdin and dispatches incoming requests.

call(method: str, args: tuple[Any, ...] = (), kwargs: dict | None = None, *, callback: Callback | None = None, block: int = 0, timeout: float | int = 0) → None[source]#

Performs an actual remote procedure call by writing a request representation (a string) to the output stream. The remote RPC instance uses method to route to the actual method to call with args and kwargs.

When callback is set, it will be called with the result of the remote call. When block is larger than 0, the calling thread is blocked until the result is received. In this case, block will be the poll interval, emulating synchronuous return value behavior. When both callback is None and block is 0 or smaller, the request is considered a notification and the remote RPC instance will not send a response.

If timeout is not zero, raise TimeoutError after timeout seconds with no response.

Parameters:

method – The method to call.
args – The positional arguments of the method.
kwargs – The keyword arguments of the method.
callback – The callback function.
block – The poll interval in seconds.
timeout – The timeout in seconds.

Returns:

None.

Raises:

TimeoutError – When the request times out.

class Spec[source]#

This class wraps methods that create JSON-RPC 2.0 compatible string representations of request, response and error objects. All methods are class members, so you might never want to create an instance of this class, but rather use the methods directly:

Spec.request("my_method", 18)  # the id is optional
# => '{"jsonrpc":"2.0","method":"my_method","id": 18}'

Spec.response(18, "some_result")
# => '{"jsonrpc":"2.0","id":18,"result":"some_result"}'

Spec.error(18, -32603)
# => '{"jsonrpc":"2.0","id":18,"error":{"code":-32603,"message":"Internal error"}}'

classmethod check_id(id: str | int | None, *, allow_empty: bool = False) → None[source]#

Value check for id entries. When allow_empty is True, id is allowed to be None. Raises a TypeError when id is neither an integer nor a string.

Parameters:

id – The id to check.
allow_empty – Whether id is allowed to be None.

Returns:

None.

Raises:

TypeError – When id is invalid.

classmethod check_method(method: str, /) → None[source]#

Value check for method entries. Raises a TypeError when method is not a string.

Parameters:: method – The method to check.
Returns:: None.
Raises:: TypeError – When method is invalid.

classmethod check_code(code: int, /) → None[source]#

Value check for code entries. Raises a TypeError when code is not an integer, or a KeyError when there is no RPCError subclass registered for that code.

Parameters:: code – The error code to check.
Returns:: None.
Raises:: TypeError – When code is invalid.

classmethod request(method: str, /, id: str | int | None = None, *, params: dict[str, Any] | None = None) → str[source]#

Creates the string representation of a request that calls method with optional params which are encoded by json.dumps. When id is None, the request is considered a notification.

Parameters:

method – The method to call.
id – The id of the request.
params – The parameters of the request.

Returns:

The request string.

Raises:

RPCInvalidRequest – When method or id are invalid.
RPCParseError – When params could not be encoded.

classmethod response(id: str | int | None, result: Any, /) → str[source]#

Creates the string representation of a respone that was triggered by a request with id. A result is required, even if it is None.

Parameters:

id – The id of the request that triggered this response.
result – The result of the request.

Returns:

The response string.

Raises:

RPCInvalidRequest – When id is invalid.
RPCParseError – When result could not be encoded.

classmethod error(id: str | int | None, code: int, *, data: Any | None = None) → str[source]#

Creates the string representation of an error that occured while processing a request with id. code must lead to a registered RPCError. data might contain additional, detailed error information and is encoded by json.dumps when set.

Parameters:

id – The id of the request that triggered this error.
code – The error code.
data – Additional error data.

Returns:

The error string.

Raises:

RPCInvalidRequest – When id or code are invalid.
RPCParseError – When data could not be encoded.

class Watchdog(rpc: RPC, name: str = 'watchdog', interval: float | int = 0.1, daemon: bool = False, start: bool = True)[source]#

This class represents a thread that watches the input stream of an RPC instance for incoming content and dispatches requests to it.

Parameters:

rpc – The RPC instance to watch.
name – The thread’s name.
interval – The polling interval of the run loop.
daemon – The thread’s daemon flag.
start – Whether to start the thread immediately.

rpc#: The RPC instance.

name#: The thread’s name.

interval#: The polling interval of the run loop.

daemon#: The thread’s daemon flag.

start() → None[source]#

Starts the thread’s activity.

Returns:: None.

stop() → None[source]#

Stops the thread’s activity.

Returns:: None.

run() → None[source]#

The main run loop of the watchdog thread. Reads the input stream of the RPC instance and dispatches incoming content to it.

Returns:: None.

exception RPCError(data: str | None = None)[source]#

Base class for RPC errors.

Parameters:: data – Additional error data.

code_range#: The range of error codes that this error class is responsible for.

code#: The error code of this error.

title#: The title of this error.

message#: The message of this error, i.e., "<title> (<code>)[, data: <data>]".

data#: Additional data of this error. Setting the data attribute will also change the message attribute.

classmethod is_code_range(code: Any) → bool[source]#

Returns True when code is a valid error code range, i.e., a tuple of two integers where the first integer is less or equal to the second integer.

Parameters:: code – The code to check.
Returns:: Whether code is a valid error code range.

register_error(cls: Type[RPCError]) → Type[RPCError][source]#

Decorator that registers a new RPC error derived from RPCError. The purpose of error registration is to have a mapping of error codes/code ranges to error classes for faster lookups during error creation.

@register_error
class MyCustomRPCError(RPCError):
    code_range = (lower_bound, upper_bound)  # both inclusive
    code = code_range[0]  # default code when used as is
    title = "My custom error"

get_error(code: int) → Type[RPCError][source]#

Returns the RPC error class that was previously registered to code. A ValueError is raised if no error class was found for code.

Parameters:: code – The error code to look up.
Returns:: The error class.
Raises:: ValueError – When no error class was found for code.

RPC errors#

exception RPCParseError(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32700, -32700)#

code: int = -32700#

title: str = 'Parse error'#

exception RPCInvalidRequest(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32600, -32600)#

code: int = -32600#

title: str = 'Invalid Request'#

exception RPCMethodNotFound(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32601, -32601)#

code: int = -32601#

title: str = 'Method not found'#

exception RPCInvalidParams(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32602, -32602)#

code: int = -32602#

title: str = 'Invalid params'#

exception RPCInternalError(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32603, -32603)#

code: int = -32603#

title: str = 'Internal error'#

exception RPCServerError(data: str | None = None)[source]#

code_range: tuple[int, int] = (-32099, -32000)#

code: int = -32099#

title: str = 'Server error'#

Project Info#

Installation#

Install simply via pip.

pip install jsonrpyc

# or with optional dev dependencies
pip install jsonrpyc[dev]

Contributing#

If you like to contribute to jsonrpyc, I’m happy to receive pull requests. Just make sure to add new test cases, run them via

> pytest tests

and check for linting and typing errors with

> mypy jsonrpyc
> flake8 jsonrpyc

Development#

Source hosted at GitHub
Report issues, questions, feature requests on GitHub Issues

jsonrpyc

Contents

jsonrpyc#

Usage#

server.py#

client.py#

API Docs#

Classes#

RPC errors#

Project Info#

Installation#

Contributing#

Development#

`server.py`#

`client.py`#