JSON-RPC

JSON-RPC client and data models.

JSONRPCPayload

Bases: typing_extensions.TypedDict

A rpc call is represented by sending a Request object to a Server.

See: https://www.jsonrpc.org/specification#request_object

JSONRPCResponse

Bases: typing_extensions.TypedDict

When a rpc call is made, the Server MUST reply with a Response.

See: https://www.jsonrpc.org/specification#response_object

JSONRPCError

Bases: typing_extensions.TypedDict

When a rpc call encounters an error.

See: https://www.jsonrpc.org/specification#error_object

JSONRPCResponseError(error: JSONRPCError)

Bases: Exception

Exception used when a rpc call encounters an error.

Source code in pytvpaint/george/client/rpc.py

def __init__(self, error: JSONRPCError) -> None:
    super().__init__(f"JSON-RPC Server error ({error['code']}): {error['message']}")

JSONRPCClient(url: str, timeout: int = 60, version: str = '2.0')

Simple JSON-RPC 2.0 client over websockets with automatic reconnection.

See: https://www.jsonrpc.org/specification#notification

Initialize a new JSON-RPC client with a WebSocket url endpoint.

Parameters:

Name	Type	Description	Default
url	str	the WebSocket url endpoint	required
timeout	int	the reconnection timeout	60
version	str	The JSON-RPC version. Defaults to "2.0".	'2.0'

Source code in pytvpaint/george/client/rpc.py

def __init__(self, url: str, timeout: int = 60, version: str = "2.0") -> None:
    """Initialize a new JSON-RPC client with a WebSocket url endpoint.

    Args:
        url: the WebSocket url endpoint
        timeout: the reconnection timeout
        version: The JSON-RPC version. Defaults to "2.0".
    """
    self.ws_handle = WebSocket()
    self.url = url
    self.rpc_id = 0
    self.timeout = timeout
    self.jsonrpc_version = version

    self.stop_ping = threading.Event()
    self.run_forever = False
    self.ping_thread: threading.Thread | None = None
    self._ping_start_time: float = 0

is_connected: bool property

Returns True if the client is connected.

connect(timeout: float | None = None) -> None

Connects to the WebSocket endpoint.

Source code in pytvpaint/george/client/rpc.py

def connect(self, timeout: float | None = None) -> None:
    """Connects to the WebSocket endpoint."""
    self.ws_handle.connect(self.url, timeout=timeout)

    if not self.ping_thread:
        self._ping_start_time = time()
        self.ping_thread = threading.Thread(
            target=self._auto_reconnect, daemon=True
        )
        self.run_forever = True
        self.ping_thread.start()

disconnect() -> None

Disconnects from the server.

Source code in pytvpaint/george/client/rpc.py

def disconnect(self) -> None:
    """Disconnects from the server."""
    self.run_forever = False
    if self.ping_thread:
        self.ping_thread.join()

    self.ws_handle.close()

increment_rpc_id() -> None

Increments the internal RPC id until it reaches sys.maxsize.

Source code in pytvpaint/george/client/rpc.py

def increment_rpc_id(self) -> None:
    """Increments the internal RPC id until it reaches `sys.maxsize`."""
    self.rpc_id = (self.rpc_id + 1) % sys.maxsize

execute_remote(method: str, params: list[JSONValueType] | None = None) -> JSONRPCResponse

Executes a remote procedure call.

Parameters:

Name	Type	Description	Default
method	str	the name of the method to be invoked	required
params	list[pytvpaint.george.client.rpc.JSONValueType] | None	the parameter values to be used during the invocation of the method. Defaults to None.	None

Raises:

Type	Description
ConnectionError	if the client is not connected
pytvpaint.george.client.rpc.JSONRPCResponseError	if there was an error server-side

Returns:

Name	Type	Description
JSONRPCResponse	pytvpaint.george.client.rpc.JSONRPCResponse	the JSON-RPC response payload

Source code in pytvpaint/george/client/rpc.py

def execute_remote(
    self,
    method: str,
    params: list[JSONValueType] | None = None,
) -> JSONRPCResponse:
    """Executes a remote procedure call.

    Args:
        method: the name of the method to be invoked
        params: the parameter values to be used during the invocation of the method. Defaults to None.

    Raises:
        ConnectionError: if the client is not connected
        JSONRPCResponseError: if there was an error server-side

    Returns:
        JSONRPCResponse: the JSON-RPC response payload
    """
    if not self.is_connected:
        raise ConnectionError(
            f"Can't send rpc message because the client is not connected to {self.url}"
        )

    payload: JSONRPCPayload = {
        "jsonrpc": self.jsonrpc_version,
        "id": self.rpc_id,
        "method": method,
        "params": params or [],
    }

    self.ws_handle.send(json.dumps(payload))
    self.increment_rpc_id()

    result = self.ws_handle.recv()
    response = cast(JSONRPCResponse, json.loads(result))

    if "error" in response:
        raise JSONRPCResponseError(response["error"])

    return response