aiohttp JSON-RPC服务器处理程序和客户端
项目描述
基于aiohttp的JSON-RPC服务器和客户端实现。
服务器示例
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)客户端示例
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())
关闭
aiohttp-jsonrpc-0.4.0.tar.gz的哈希值
| 算法 | 哈希摘要 | |
|---|---|---|
| SHA256 | 3bd05afe985c4d50e500dcf2f9fc00a61a24eb6036857f49372d19cd11416351 |
|
| MD5 | 5288dd81a0b3436243bfb5cb43874a3c |
|
| BLAKE2b-256 | 8a65dbcc69c5d6e952c8c2bfdbfe3f9783f4e4c1816fa93916a045a52c9d5506 |