aiohttp JSON-RPC server handler and client
Project description
JSON-RPC server and client implementation based on aiohttp.
Server example
from aiohttp import web
from aiohttp_jsonrpc import handler
class JSONRPCExample(handler.JSONRPCView):
def rpc_test(self):
return None
def rpc_args(self, *args):
return len(args)
def rpc_kwargs(self, **kwargs):
return len(kwargs)
def rpc_args_kwargs(self, *args, **kwargs):
return len(args) + len(kwargs)
def rpc_exception(self):
raise Exception("YEEEEEE!!!")
def rpc_test_notification(self):
print("Notification received")
app = web.Application()
app.router.add_route('*', '/', JSONRPCExample)
if __name__ == "__main__":
import logging
logging.basicConfig(level=logging.INFO)
web.run_app(app, print=logging.info)
Client example
import asyncio
from aiohttp_jsonrpc.client import ServerProxy
loop = asyncio.get_event_loop()
client = ServerProxy("http://127.0.0.1:8080/", loop=loop)
async def main():
print(await client.test())
# Or via __getitem__
method = client['args']
notification = client.create_notification("test_notification")
print(await method(1, 2, 3))
await notification()
results = await client.batch(
client['test'],
client['test'].prepare(),
client['args'].prepare(1, 2, 3),
client['not_found'].prepare(1, 2, 3),
# notify with params
notification.prepare(),
# notification without params
notification,
)
print(results)
client.close()
if __name__ == "__main__":
loop.run_until_complete(main())
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
aiohttp-jsonrpc-0.4.0.tar.gz
(8.3 kB
view hashes)
Built Distribution
Close
Hashes for aiohttp_jsonrpc-0.4.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cdd34a5f6c689653c75ba68146422990a0fce6c830d88c8fbf69de84dab2566f |
|
MD5 | d4d863c95d9c5cabf35020f2f870d277 |
|
BLAKE2b-256 | e264b9fee2d2ad6ad3c1555cbf86f34336e41532047aa48028b21fc20c0a32f1 |