Client#

Misc#

JSON-RPC client.

class pjrpc.client.AbstractAsyncClient(*, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=(), request_content_type='application/json', response_content_types=('application/json', 'application/json-rpc'))[source]#

Bases: ABC

Abstract asynchronous JSON-RPC client.

Parameters:

id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
middlewares (Iterable[AsyncMiddleware]) – client reqeust middlewares
request_content_type (str) –
response_content_types (Iterable[str]) –

class Proxy(client)[source]#

Bases: object

Proxy object. Provides syntactic sugar to make method call using dot notation.

Parameters:: client (AbstractAsyncClient) – JSON-RPC client instance

property proxy: Proxy#: Client proxy object.

batch()[source]#

Client batch wrapper.

Return type:: AsyncGenerator[Batch, None]

async notify(method, *args, **kwargs)[source]#

Makes a notification request

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Return type:

None

async call(method, *args, **kwargs)[source]#

Makes a JSON-RPC call.

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Returns:

response result

Return type:

Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef(‘JsonT’)], tuple[ForwardRef(‘JsonT’), …], dict[str, ForwardRef(‘JsonT’)]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef(‘JsonT’)], tuple[ForwardRef(‘JsonT’), …], dict[str, ForwardRef(‘JsonT’)]], …], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef(‘JsonT’)], tuple[ForwardRef(‘JsonT’), …], dict[str, ForwardRef(‘JsonT’)]]]]

class pjrpc.client.AbstractClient(*, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=(), request_content_type='application/json', response_content_types=('application/json', 'application/json-rpc'))[source]#

Bases: ABC

Abstract synchronous JSON-RPC client.

Parameters:

id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
middlewares (Iterable[Middleware]) – client reqeust middlewares
request_content_type (str) –
response_content_types (Iterable[str]) –

class Proxy(client)[source]#

Bases: object

Proxy object. Provides syntactic sugar to make method call using dot notation.

Parameters:: client (AbstractClient) – JSON-RPC client instance

property proxy: Proxy#: Client proxy object.

batch()[source]#

Client batch wrapper.

Return type:: Generator[Batch, None, None]

notify(method, *args, **kwargs)[source]#

Makes a notification request

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Return type:

None

call(method, *args, **kwargs)[source]#

Makes a JSON-RPC call.

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Returns:

response result

Return type:

class pjrpc.client.AsyncMiddleware(*args, **kwargs)[source]#

Bases: Protocol

Asynchronous JSON-RPC client middleware.

class pjrpc.client.Batch(id_gen_impl)[source]#

Bases: object

Batch object. Provides syntactic sugar to send batch requests.

Parameters:: id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) –

class Proxy(batch)[source]#

Bases: object

Proxy object. Provides syntactic sugar to make method call using dot notation.

Parameters:: batch (Batch) – batch object

property proxy: Proxy#: Client proxy object.

property requests: list[pjrpc.common.request.Request]#: Batch requests.

notify(method, *args, **kwargs)[source]#

Makes a notification request

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Return type:

None

call(method, *args, **kwargs)[source]#

Makes a JSON-RPC call.

Parameters:

method (str) – method name
args (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method positional arguments
kwargs (Union[str, int, float, bool, None, list[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]], tuple[Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]], ...], dict[str, Union[str, int, float, bool, NoneType, list[ForwardRef('JsonT')], tuple[ForwardRef('JsonT'), ...], dict[str, ForwardRef('JsonT')]]]]) – method named arguments

Returns:

response result

Return type:

None

set_response(response)[source]#

Sets batch response

Parameters:: response (Optional[BatchResponse]) –
Return type:: None

get_response()[source]#

Returns a batch response.

Return type:: Optional[BatchResponse]

get_results()[source]#

Returns the batch results preserving requests order (skipping notification request).

Return type:: Iterable[Any]

exception pjrpc.client.JsonRpcError(code, message, data=UnsetType.UNSET)[source]#

Bases: JsonRpcError

Client JSON-RPC error.

Parameters:

code (int) –
message (str) –
data (Union[UnsetType, Any]) –

Return type:

None

class pjrpc.client.Middleware(*args, **kwargs)[source]#

Bases: Protocol

JSON-RPC client middleware.

Backends#

requests#

class pjrpc.client.backend.requests.RequestArgs[source]#: Bases: TypedDict

class pjrpc.client.backend.requests.Client(url, *, session=None, raise_for_status=True, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=())[source]#

Bases: AbstractClient

Requests library client backend.

Parameters:

url (str) – url to be used as JSON-RPC endpoint.
session (Optional[Session]) – custom session to be used instead of requests.Session
id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
raise_for_status (bool) –
middlewares (Iterable[Middleware]) –

send(request: Request, **kwargs: Any) → Optional[Response][source]#

send(request: BatchRequest, **kwargs: Any) → Optional[BatchResponse]

Sends a JSON-RPC request.

Parameters:

request – request instance
kwargs – additional client request argument

Returns:

response instance or None if the request is a notification

close()[source]#

Closes the current http session.

Return type:: None

httpx#

class pjrpc.client.backend.httpx.RequestArgs[source]#: Bases: TypedDict

class pjrpc.client.backend.httpx.Client(url, *, http_client=None, raise_for_status=True, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=())[source]#

Bases: AbstractClient

httpx library sync client backend.

Parameters:

url (str) – url to be used as JSON-RPC endpoint.
http_client (Optional[Client]) – custom client to be used instead of httpx.Client
raise_for_status (bool) – should ClientResponse.raise_for_status() be called automatically
id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
middlewares (Iterable[Middleware]) –

send(request: Request, **kwargs: Any) → Optional[Response][source]#

send(request: BatchRequest, **kwargs: Any) → Optional[BatchResponse]

Sends a JSON-RPC request.

Parameters:

request – request instance
kwargs – additional client request argument

Returns:

response instance or None if the request is a notification

close()[source]#

Closes the current http session.

Return type:: None

class pjrpc.client.backend.httpx.AsyncClient(url, *, http_client=None, raise_for_status=True, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=())[source]#

Bases: AbstractAsyncClient

httpx library async client backend.

Parameters:

url (str) – url to be used as JSON-RPC endpoint.
http_client (Optional[AsyncClient]) – custom client to be used instead of httpx.AsyncClient
raise_for_status (bool) – should ClientResponse.raise_for_status() be called automatically
id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
middlewares (Iterable[AsyncMiddleware]) –

async send(request: Request, **kwargs: Any) → Optional[Response][source]#

async send(request: BatchRequest, **kwargs: Any) → Optional[BatchResponse]

Sends a JSON-RPC request.

Parameters:

request – request instance
kwargs – additional client request argument

Returns:

response instance or None if the request is a notification

async close()[source]#

Closes current http session.

Return type:: None

aiohttp#

class pjrpc.client.backend.aiohttp.RequestArgs[source]#: Bases: TypedDict

class pjrpc.client.backend.aiohttp.Client(url, *, session=None, raise_for_status=True, id_gen_impl=<function sequential>, error_cls=<class 'pjrpc.client.exceptions.JsonRpcError'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.common.encoder.JSONEncoder'>, json_decoder=None, middlewares=())[source]#

Bases: AbstractAsyncClient

Aiohttp library client backend.

Parameters:

url (str) – url to be used as JSON-RPC endpoint
session (Optional[ClientSession]) – custom session to be used instead of aiohttp.ClientSession
raise_for_status (bool) – should ClientResponse.raise_for_status() be called automatically
id_gen_impl (Callable[[...], Generator[Union[str, int], None, None]]) – identifier generator
error_cls (type[pjrpc.client.exceptions.JsonRpcError]) – JSON-RPC error base class
json_loader (Callable[[...], Any]) – json loader
json_dumper (Callable[[...], str]) – json dumper
json_encoder (type[pjrpc.common.encoder.JSONEncoder]) – json encoder
json_decoder (Optional[JSONDecoder]) – json decoder
middlewares (Iterable[AsyncMiddleware]) –

async send(request: Request, **kwargs: Any) → Optional[Response][source]#

async send(request: BatchRequest, **kwargs: Any) → Optional[BatchResponse]

Sends a JSON-RPC request.

Parameters:

request – request instance
kwargs – additional client request argument

Returns:

response instance or None if the request is a notification

async close()[source]#

Closes current http session.

Return type:: None

aio-pika#

Retry#

class pjrpc.client.retry.Backoff(attempts, jitter=<function Backoff.<lambda>>)[source]#

Bases: object

JSON-RPC request retry strategy.

Parameters:

attempts (int) – retries number
jitter (Callable[[int], float]) – retry delay jitter generator

class pjrpc.client.retry.PeriodicBackoff(attempts, jitter=<function Backoff.<lambda>>, interval=1.0)[source]#

Bases: Backoff

Periodic request retry strategy.

Parameters:

interval (float) – retry delay
attempts (int) –
jitter (Callable[[int], float]) –

class pjrpc.client.retry.ExponentialBackoff(attempts, jitter=<function Backoff.<lambda>>, base=1.0, factor=2.0, max_value=None)[source]#

Bases: Backoff

Exponential request retry strategy.

Parameters:

base (float) – exponentially growing delay base
factor (float) – exponentially growing delay factor (multiplier)
max_value (Optional[float]) – delay max value
attempts (int) –
jitter (Callable[[int], float]) –

class pjrpc.client.retry.FibonacciBackoff(attempts, jitter=<function Backoff.<lambda>>, multiplier=1.0, max_value=1.0)[source]#

Bases: Backoff

Fibonacci request retry strategy.

Parameters:

multiplier (float) – fibonacci interval sequence multiplier
max_value (float) – delay max value
attempts (int) –
jitter (Callable[[int], float]) –

class pjrpc.client.retry.RetryStrategy(backoff, codes=None, exceptions=None)[source]#

Bases: object

JSON-RPC request retry strategy.

Parameters:

backoff (Backoff) – backoff delay generator
codes (Optional[set[int]]) – JSON-RPC response codes receiving which the request will be retried
exceptions (Optional[set[type[Exception]]]) – exceptions catching which the request will be retried

Integrations#

pytest client library integration. Implements some utilities for mocking out pjrpc library clients.

class pjrpc.client.integrations.pytest.Match(endpoint, version, method_name, once, callback, **response_data)[source]#

Bases: object

Match object. Incorporates request matching information.

Parameters:

endpoint (str) –
version (str) –
method_name (str) –
once (bool) –
callback (Optional[Callable[[...], Any]]) –
response_data (Any) –

class pjrpc.client.integrations.pytest.PjRpcMocker(target, mocker=<module 'unittest.mock' from '/home/docs/.asdf/installs/python/3.12.10/lib/python3.12/unittest/mock.py'>, passthrough=False, dumps=<function dumps>)[source]#

Bases: object

Synchronous JSON-RPC client mocker.

Parameters:

target (str) – method to be mocked
mocker (module) – mocking package
passthrough (bool) – pass not mocked requests to the original method
dumps (Callable[[...], Any]) – response dump function

property calls: Dict[str, Dict[Tuple[str, str], Mock]]#: Dictionary of JSON-PRC method calls.

add(endpoint, method_name, result=UnsetType.UNSET, error=UnsetType.UNSET, id=None, version='2.0', once=False, callback=None)[source]#

Appends response patch. If the same method patch already exists they will be used in a round-robin way.

Parameters:

endpoint (str) – request endpoint
method_name (str) – method name
result (Union[UnsetType, Any]) – patched result
error (Union[UnsetType, Any]) – patched error
id (Optional[Union[str, int]]) – patched request id
version (str) – patched request version
once (bool) – if True the patch will be deleted after the first call
callback (Optional[Callable[[...], Any]]) – patched request callback

Return type:

None

replace(endpoint, method_name, result=UnsetType.UNSET, error=UnsetType.UNSET, id=None, version='2.0', once=False, callback=None, idx=0)[source]#

Replaces a previously added response patch by a new one.

Parameters:

endpoint (str) – request endpoint
method_name (str) – method name
result (Union[UnsetType, Any]) – patched result
error (Union[UnsetType, Any]) – patched error
id (Optional[Union[str, int]]) – patched request id
version (str) – patched request version
once (bool) – if True the patch will be deleted after the first call
callback (Optional[Callable[[...], Any]]) – patched request callback
idx (int) – patch index (if there are more than one)

Return type:

None

remove(endpoint, method_name=None, version='2.0')[source]#

Removes a previously added response patch.

Parameters:

endpoint (str) – request endpoint
method_name (Optional[str]) – method name
version (str) – JSON-RPC request version

Returns:

removed response patch

Return type:

Union[Dict[Tuple[str, str], List[Match]], List[Match]]

reset()[source]#

Removes all added matches and reset call statistics.

Return type:: None

start()[source]#

Activates a patcher.

Return type:: Any

stop()[source]#

Stop an active patcher.

Return type:: None

pjrpc.client.integrations.pytest_aiohttp.pjrpc_aiohttp_mocker()[source]#

Aiohttp client mocking fixture.

Return type:: Generator[PjRpcMocker, None, None]

pjrpc.client.integrations.pytest_requests.pjrpc_requests_mocker()[source]#

Requests client mocking fixture.

Return type:: Generator[PjRpcMocker, None, None]