Warning
This is the documentation for version 4, released September 1, 2021. Read about the changes in version 4.
Welcome to jsonrpcclient’s documentation!¶
Generate JSON-RPC requests and parse responses in Python.
pip install jsonrpcclient
Generating requests¶
The request function¶
Generate a request with the request
function:
>>> from jsonrpcclient import request
>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 1}
Ids¶
Subsequent calls increment the id
:
>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 2}
>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 3}
Use an explicit id
:
>>> request("ping", id="foo")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": "foo"}
Or generate a different type of id
:
>>> from jsonrpcclient import request_hex, request_random, request_uuid
>>> request_hex("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "1"}
>>> request_random("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "qzsib147"}
>>> request_uuid("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "45480a2f-069c-42aa-a67f-f6fdd83d6026"}
Parameters¶
Pass params
to include parameters in the payload. This should be either a
tuple for positional arguments, or dict for keyword arguments.
>>> request("ping", params=(1,))
{"jsonrpc": "2.0", "method": "ping", "2.0", "params": [1], "id": 4}
>>> request("ping", params={"key": "val"})
{"jsonrpc": "2.0", "method": "ping", "2.0", "params": {"key": "val"}, "id": 5}
JSON requests¶
If you need the request serialized to a string, use request_json
:
>>> from jsonrpcclient import request_json
>>> request_json("foo")
'{"jsonrpc": "2.0", "method": "foo", "id": 6}'
You can also use request_json_hex etc., for the other id types.
Batch requests¶
>>> import json
>>> json.dumps([request("foo") for _ in range(3)])
'[{"jsonrpc": "2.0", "method": "foo", "id": 7}, {"jsonrpc": "2.0", "method": "foo", "id": 8}, {"jsonrpc": "2.0", "method": "foo", "id": 9}]'
Notifications¶
Use the notification
function instead of request
:
>>> from jsonrpcclient import notification
>>> notification("ping")
{"jsonrpc": "2.0", "method": "ping"}
Similar to request_json
, notification_json
will give you the notification
as a JSON string.
>>> from jsonrpcclient import notification_json
>>> notification_json("ping")
'{"jsonrpc": "2.0", "method": "ping"}'
Parsing responses¶
The library includes a parse
function which turns a deserialized response
into a nice namedtuple.
>>> parse({"jsonrpc": "2.0", "result": "pong", "id": 1})
Ok(result='pong', id=1)
>>> parse({"jsonrpc": "2.0", "error": {"code": 1, "message": "There was an error", "data": None}, "id": 1})
Error(code=1, message='There was an error', data=None, id=1)
If you have a string, use parse_json
.
>>> parse_json('{"jsonrpc": "2.0", "result": "pong", "id": 1}')
Ok(result='pong', id=1)
To use the result, in Python versions prior to 3.10:
from jsonrpcclient import Error, Ok
parsed = parse(response)
if isinstance(parsed, Ok):
print(parsed.result)
else:
logging.error(parsed.message)
In Python 3.10+, use pattern matching:
match parse(response):
case Ok(result, id):
print(result)
case Error(code, message, data, id):
logging.error(message)
Examples¶
Contents
aiohttp¶
import asyncio
import logging
from aiohttp import ClientSession
from jsonrpcclient import Ok, request, parse
async def main():
async with ClientSession() as session:
async with session.post(
"http://localhost:5000", json=request("ping")
) as response:
parsed = parse(await response.json())
if isinstance(parsed, Ok):
print(parsed.result)
else:
logging.error(parsed.message)
asyncio.get_event_loop().run_until_complete(main())
See blog post.
Requests¶
from jsonrpcclient import request, parse, Ok
import logging
import requests
response = requests.post("http://localhost:5000/", json=request("ping"))
parsed = parse(response.json())
if isinstance(parsed, Ok):
print(parsed.result)
else:
logging.error(parsed.message)
Websockets¶
import asyncio
import logging
from jsonrpcclient import Ok, parse_json, request_json
import websockets
async def main():
async with websockets.connect("ws://localhost:5000") as ws:
await ws.send(request_json("ping"))
response = parse_json(await ws.recv())
if isinstance(response, Ok):
print(response.result)
else:
logging.error(response.message)
asyncio.get_event_loop().run_until_complete(main())
See blog post.
ZeroMQ¶
from jsonrpcclient import request_json, parse_json, Ok
import logging
import zmq
socket = zmq.Context().socket(zmq.REQ)
socket.connect("tcp://localhost:5000")
socket.send_string(request_json("ping"))
response = parse_json(socket.recv().decode())
if isinstance(response, Ok):
print(response.result)
else:
logging.error(response.message)
See blog post.
FAQ¶
How to use a different json library?¶
The request_json
function is simply json.dumps
after request
, and the
parse_json
function is simply parse
after json.loads
.
So here’s how one could write their own, using a different json library (ujson here):
from jsonrpcclient import request, parse
from jsonrpcclient.utils import compose
import ujson
parse_json = compose(parse, ujson.loads)
request_json = compose(ujson.dumps, request)