A simple JSON-RPC for aiohttp
Project description
aiohttp-rpc
A library for a simple integration of the JSON-RPC 2.0 protocol to a Python application using aiohttp. The motivation is to provide a simple, fast and reliable way to integrate the JSON-RPC 2.0 protocol into your application on the server and/or client side.
The library has only one dependency:
- aiohttp - Async http client/server framework
Table Of Contents
Installation
pip
pip install aiohttp-rpc
Usage
HTTP Server Example
from aiohttp import web
import aiohttp_rpc
@aiohttp_rpc.rpc_method()
def echo(*args, **kwargs):
return {
'args': args,
'kwargs': kwargs,
}
# If the function has rpc_request in arguments, then it is automatically passed
async def ping(rpc_request):
return 'pong'
if __name__ == '__main__':
aiohttp_rpc.rpc_server.add_methods([
('', ping,),
])
app = web.Application()
app.router.add_routes([
web.post('/rpc', aiohttp_rpc.rpc_server.handle_http_request),
])
web.run_app(app, host='0.0.0.0', port=8080)
HTTP Client Example
import aiohttp_rpc
import asyncio
async def run():
async with aiohttp_rpc.JsonRpcClient('http://0.0.0.0:8080/rpc') as rpc:
print(await rpc.ping())
print(await rpc.echo(a=4, b=6))
print(await rpc.call('echo', a=4, b=6))
print(await rpc.notify('echo', 1, 2, 3))
print(await rpc.echo(1, 2, 3))
print(await rpc.batch([
['echo', 2],
'echo2',
'ping',
]))
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
Integration
The purpose of this library is to simplify life, and not vice versa. And so, when you start adding existing functions, some problems may arise.
Existing functions can return objects that are not serialized, but this is easy to fix.
You can add own json_serialize
:
async def get_user_by_uuid(user_uuid: typing.Union[str, uuid.uuid4]) -> typing.Optional[User]:
pass
def json_serialize_unknown_value(value) -> typing.Any:
if isinstance(value, User):
return {
'id': value.id,
'uuid': str(value.uuid),
'username': value.username,
'email': value.email,
}
return repr(value)
rpc_server = aiohttp_rpc.JsonRpcServer(
json_serialize=partial(json.dumps, default=json_serialize_unknown_value),
)
rpc_server.add_methods((
get_user_by_uuid,
))
"""
Response:
{
"id": null,
"jsonrpc": "2.0",
"result": {
"id": 2510,
"uuid": "600d57b3-dda8-43d0-af79-3e81dbb344fa",
"username": "Mike",
"email": "some@mail.com"
}
}
"""
If you need to replace the function arguments, then you can use middleware.
Middleware
Middleware is used for request/response processing.
import aiohttp_rpc
class TokenMiddleware(aiohttp_rpc.BaseJsonRpcMiddleware):
async def __call__(self, request: aiohttp_rpc.JsonRpcRequest) -> aiohttp_rpc.JsonRpcResponse:
if request.http_request and request.http_request.headers.get('X-App-Token') != 'qwerty':
raise exceptions.InvalidRequest('Invalid token')
return await self.get_response(request)
rpc_server = aiohttp_rpc.JsonRpcServer(middlewares=[
aiohttp_rpc.ExceptionMiddleware,
TokenMiddleware,
])
To process web.Request
/web.Response
, wrap function aiohttp_rpc.JsonServer.handle_http_request
:
def my_handle_http_request(request):
# something
# for example, authorization
return aiohttp_rpc.rpc_server.handle_http_request
...
app = web.Application()
app.router.add_routes((
web.post('/rpc', my_handle_http_request),
))
Or add `middleware` in [aiohttp](https://github.com/aio-libs/aiohttp).
....
WebSockets
WS Server Example
from aiohttp import web
import aiohttp_rpc
async def ping(rpc_request):
return 'pong'
if __name__ == '__main__':
rpc_server = aiohttp_rpc.WsJsonRpcServer(
middlewares=aiohttp_rpc.middlewares.DEFAULT_MIDDLEWARES,
)
rpc_server.add_method(ping)
app = web.Application()
app.router.add_routes([
web.get('/rpc', rpc_server.handle_http_request),
])
web.run_app(app, host='0.0.0.0', port=8080)
WS Client Example
import aiohttp_rpc
import asyncio
async def run():
async with aiohttp_rpc.WsJsonRpcClient('http://0.0.0.0:8080/rpc') as rpc:
print(await rpc.ping())
print(await rpc.notify('ping'))
print(await rpc.batch([
['echo', 2],
'echo2',
'ping',
]))
loop = asyncio.get_event_loop()
loop.run_until_complete(run())
API Reference
server
-
class aiohttp_rpc.JsonRpc(BaseJsonRpcServer)
def __init__(self, *, son_serialize = aiohttp_rpc.utils.json_serialize, middlewares: typing.Iterable = (), methods = None)
def add_method(self, method, *, replace = False) -> aiohttp_rpc.JsonRpcMethod
def add_methods(self, methods, replace = False) -> typing.List[aiohttp_rpc.JsonRpcMethod]:
def get_methods(self) -> dict
async def handle_http_request(self, http_request: web.Request) -> web.Response
:
-
class aiohttp_rpc.JsonRpcClient(BaseJsonRpcClient)
client
class aiohttp_rpc.WsJsonRpcServer(BaseJsonRpcServer)
class aiohttp_rpc.WsJsonRpcClient(BaseJsonRpcClient)
class UnlinkedResults
protocol
class aiohttp_rpc.JsonRpcRequest
class aiohttp_rpc.JsonRpcResponse
class aiohttp_rpc.JsonRpcMethod
decorators
def rpc_method(prefix = '', *, rpc_server = default_rpc_server, custom_name = None, add_extra_args = True)
errors
class JsonRpcError(RuntimeError)
class ServerError(JsonRpcError)
class ParseError(JsonRpcError)
class InvalidRequest(JsonRpcError)
class MethodNotFound(JsonRpcError)
class InvalidParams(JsonRpcError)
class InternalError(JsonRpcError)
DEFAULT_KNOWN_ERRORS
middlewares
class BaseJsonRpcMiddleware(abc.ABC)
class ExceptionMiddleware(BaseJsonRpcMiddleware)
class ExtraArgsMiddleware(BaseJsonRpcMiddleware)
More examples
The library allows you to add methods in many ways.
import aiohttp_rpc
async def ping(rpc_request): return 'pong'
async def ping_1(rpc_request): return 'pong 1'
async def ping_2(rpc_request): return 'pong 2'
async def ping_3(rpc_request): return 'pong 3'
rpc_server = aiohttp_rpc.JsonRpcServer()
rpc_server.add_method(ping) # 'ping'
rpc_server.add_method(['', ping_1]) # 'ping_1'
rpc_server.add_method(['super', ping_1]) # 'super__ping_1'
rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('super', ping_2)) # 'super__ping_2'
rpc_server.add_method(aiohttp_rpc.JsonRpcMethod('', ping_2, custom_name='super_ping')) # 'super__super_ping'
# Replace method
rpc_server.add_method(['', ping_1], replace=True) # 'ping_1'
rpc_server.add_methods([ping_1, ping_2], replace=True) # 'ping_1', 'ping_2'
rpc_server.add_methods([['new', ping_2], ping_3]) # 'new__ping2', 'ping_3'
License
MIT
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
Built Distribution
Hashes for aiohttp_rpc-0.5.1-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | d1e19039183d0a572c2b5ea35109102197562bc1ab3f938c3c4c926b9439f07c |
|
MD5 | e95cad292f995d777b78c5458c50b945 |
|
BLAKE2b-256 | dd3429b2fd0ab1c7a0df8ed7294c892a6d5b15780c954910e29aee4f65614c48 |