-{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"Home","text":"<p>Generate JSON-RPC requests and parse responses in Python.</p> <pre><code>pip install jsonrpcclient\n</code></pre>"},{"location":"#documentation","title":"Documentation","text":"<ul> <li>Requests</li> <li>Responses</li> <li>Faq</li> <li>Examples</li> </ul>"},{"location":"examples/","title":"Examples","text":""},{"location":"examples/#aiohttp","title":"aiohttp","text":"<pre><code>import asyncio\nimport logging\n\nfrom aiohttp import ClientSession\n\nfrom jsonrpcclient import Error, Ok, parse, request\n\n\nasync def main() -> None:\n \"\"\"Handle async request\"\"\"\n async with ClientSession() as session:\n async with session.post(\n \"http://localhost:5000\", json=request(\"ping\")\n ) as response:\n parsed = parse(await response.json())\n if isinstance(parsed, Ok):\n print(parsed.result)\n elif isinstance(parse, Error):\n logging.error(parsed.message)\n\n\nasyncio.get_event_loop().run_until_complete(main())\n</code></pre> <p>See blog post.</p>"},{"location":"examples/#requests","title":"Requests","text":"<pre><code>import logging\n\nimport requests\n\nfrom jsonrpcclient import Error, Ok, parse, request\n\nresponse = requests.post(\"http://localhost:5000/\", json=request(\"ping\"), timeout=10)\nparsed = parse(response.json())\nif isinstance(parsed, Ok):\n print(parsed.result)\nelif isinstance(parsed, Error):\n logging.error(parsed.message)\n</code></pre>"},{"location":"examples/#websockets","title":"Websockets","text":"<pre><code>import asyncio\nimport logging\n\nfrom websockets.client import connect\n\nfrom jsonrpcclient import Error, Ok, parse_json, request_json\n\n\nasync def main() -> None:\n \"\"\"Handle request\"\"\"\n async with connect(\"ws://localhost:5000\") as socket:\n await socket.send(request_json(\"ping\"))\n response = parse_json(await socket.recv())\n\n if isinstance(response, Ok):\n print(response.result)\n elif isinstance(response, Error):\n logging.error(response.message)\n\n\nasyncio.get_event_loop().run_until_complete(main())\n</code></pre> <p>See blog post.</p>"},{"location":"examples/#zeromq","title":"ZeroMQ","text":"<pre><code>import logging\n\nimport zmq\n\nfrom jsonrpcclient import Ok, parse_json, request_json\n\nsocket = zmq.Context().socket(zmq.REQ)\nsocket.connect(\"tcp://localhost:5000\")\nsocket.send_string(request_json(\"ping\"))\nresponse = parse_json(socket.recv().decode())\nif isinstance(response, Ok):\n print(response.result)\nelse:\n logging.error(response.message)\n</code></pre> <p>See blog post.</p>"},{"location":"faq/","title":"FAQ","text":""},{"location":"faq/#how-to-use-a-different-json-library","title":"How to use a different json library?","text":"<p>The <code>request_json</code> function is simply <code>json.dumps</code> after <code>request</code>, and the <code>parse_json</code> function is simply <code>parse</code> after <code>json.loads</code>.</p> <p>So here's how one could write their own, using a different json library (ujson here):</p> <pre><code>from jsonrpcclient import request, parse\nfrom jsonrpcclient.utils import compose\nimport ujson\n\nparse_json = compose(parse, ujson.loads)\nrequest_json = compose(ujson.dumps, request)\n</code></pre>"},{"location":"requests/","title":"Generating requests","text":""},{"location":"requests/#the-request-function","title":"The request function","text":"<p>Generate a request with the <code>request</code> function:</p> <pre><code>>>> from jsonrpcclient import request\n>>> request(\"ping\")\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"id\": 1}\n</code></pre>"},{"location":"requests/#ids","title":"Ids","text":"<p>Subsequent calls increment the <code>id</code>: <pre><code>>>> request(\"ping\")\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"id\": 2}\n>>> request(\"ping\")\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"id\": 3}\n</code></pre></p> <p>Use an explicit <code>id</code>: <pre><code>>>> request(\"ping\", id=\"foo\")\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"id\": \"foo\"}\n</code></pre></p> <p>Or generate a different type of <code>id</code>: <pre><code>>>> from jsonrpcclient import request_hex, request_random, request_uuid\n>>> request_hex(\"foo\")\n{\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": \"1\"}\n>>> request_random(\"foo\")\n{\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": \"qzsib147\"}\n>>> request_uuid(\"foo\")\n{\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": \"45480a2f-069c-42aa-a67f-f6fdd83d6026\"}\n</code></pre></p>"},{"location":"requests/#parameters","title":"Parameters","text":"<p>Pass <code>params</code> to include parameters in the payload. This should be either a tuple for positional arguments, or dict for keyword arguments.</p> <pre><code>>>> request(\"ping\", params=(1,))\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"params\": [1], \"id\": 4}\n>>> request(\"ping\", params={\"key\": \"val\"})\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\", \"2.0\", \"params\": {\"key\": \"val\"}, \"id\": 5}\n</code></pre>"},{"location":"requests/#json-requests","title":"JSON requests","text":"<p>If you need the request serialized to a string, use <code>request_json</code>: <pre><code>>>> from jsonrpcclient import request_json\n>>> request_json(\"foo\")\n'{\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": 6}'\n</code></pre></p> <p>You can also use request_json_hex etc., for the other id types.</p>"},{"location":"requests/#batch-requests","title":"Batch requests","text":"<pre><code>>>> import json\n>>> json.dumps([request(\"foo\") for _ in range(3)])\n'[{\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": 7}, {\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": 8}, {\"jsonrpc\": \"2.0\", \"method\": \"foo\", \"id\": 9}]'\n</code></pre>"},{"location":"requests/#notifications","title":"Notifications","text":"<p>Use the <code>notification</code> function instead of <code>request</code>: <pre><code>>>> from jsonrpcclient import notification\n>>> notification(\"ping\")\n{\"jsonrpc\": \"2.0\", \"method\": \"ping\"}\n</code></pre></p> <p>Similar to <code>request_json</code>, <code>notification_json</code> will give you the notification as a JSON string.</p> <pre><code>>>> from jsonrpcclient import notification_json\n>>> notification_json(\"ping\")\n'{\"jsonrpc\": \"2.0\", \"method\": \"ping\"}'\n</code></pre>"},{"location":"responses/","title":"Parsing responses","text":"<p>The library includes a <code>parse</code> function which turns a deserialized response into a nice namedtuple.</p> <pre><code>>>> parse({\"jsonrpc\": \"2.0\", \"result\": \"pong\", \"id\": 1})\nOk(result='pong', id=1)\n>>> parse({\"jsonrpc\": \"2.0\", \"error\": {\"code\": 1, \"message\": \"There was an error\", \"data\": None}, \"id\": 1})\nError(code=1, message='There was an error', data=None, id=1)\n</code></pre> <p>If you have a string, use <code>parse_json</code>. <pre><code>>>> parse_json('{\"jsonrpc\": \"2.0\", \"result\": \"pong\", \"id\": 1}')\nOk(result='pong', id=1)\n</code></pre></p> <p>To use the result, in Python versions prior to 3.10: <pre><code>from jsonrpcclient import Error, Ok\nparsed = parse(response)\nif isinstance(parsed, Ok):\n print(parsed.result)\nelif isinstance(parse, Error):\n logging.error(parsed.message)\n</code></pre></p> <p>In Python 3.10+, use pattern matching: <pre><code>match parse(response):\n case Ok(result, id):\n print(result)\n case Error(code, message, data, id):\n logging.error(message)\n</code></pre></p>"}]}
0 commit comments