aiohttp-jrpc 0.2.0

JSON RPC protocol for aiohttp.web (http server for asyncio)

aiohttp_jrpc

jsonrpc protocol implementation for aiohttp.web.

Example server

import asyncio
from aiohttp import web
from aiohttp_jrpc import Service, JError, jrpc_errorhandler_middleware

SCH = {
    "type": "object",
    "properties": {
        "data": {"type": "string"},
    },
}

async def custom_errorhandler_middleware(app, handler):
    async def middleware(request):
        try:
            return (await handler(request))
        except Exception:
            """ Custom errors: -32000 to -32099 """
            return JError().custom(-32000, "Example error")
    return middleware

class MyJRPC(Service):
    @Service.valid(SCH)
    async def hello(self, ctx, data):
        if data["data"] == "hello":
            return {"status": "hi"}
        return {"status": data}

    async def error(self, ctx, data):
        raise Exception("Error which will catch middleware")

    async def no_valid(self, ctx, data):
        """ Method without validation incommig data """
        return {"status": "ok"}

async def init(loop):
    app = web.Application(loop=loop, middlewares=[jrpc_errorhandler_middleware])
    #app = web.Application(loop=loop, middlewares=[custom_errorhandler_middleware])
    app.router.add_route('POST', "/api", MyJRPC)

    srv = await loop.create_server(app.make_handler(),
                                        "127.0.0.1", 8080)
    print("Server started at http://127.0.0.1:8080")
    return srv

loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

Example client

import asyncio
import aiohttp
from aiohttp_jrpc import Client, InvalidResponse

URL = 'http://localhost:8080/api'

async def rpc_call():
    try:
        async with aiohttp.ClientSession() as client:
            Remote = Client(client, URL)
            rsp = await Remote.call('hello', {'data': 'hello'})
            return rsp
    except InvalidResponse as err:
        return err
    except Exception as err:
        return err
    return False

loop = asyncio.get_event_loop()
content = loop.run_until_complete(rpc_call())
print(content.result)
loop.close()

License

aiohttp_jrpc BSD license.

CHANGES

0.2.0 (2026-07-16)

• Added new tests

• async functions in classes after removing coroutine decorator

• Use enum with error messages

• Updated to work on latest python and aiohttp

0.1.0 (2016-02-20)

• Added client and tests

• Changed BSD v3 to BSD v2 license

0.0.3 (2015-10-27)

• Fix messages of protocol errors

• Fix tests and add tests for custom errors

• Fix example bugs

• Added custom middleware to example for handle errors

0.0.2 (2015-10-22)

• Added middleware to catch exceptions

• Testing internal error

0.0.1 (2015-10-18)

• Init release

Credits

aiohttp_jrpc is written by Veniamin Gvozdikov.

Contributors

• André Roth

• latyas(懒)

Please add yourself here alphabetically when you submit your first pull request.