Asyncio support

The library has full support of asyncio, though you need to be aware it has some limitations.


A very basic example on how to do raw API call:

import asyncio
from grapheneapi.aio.websocket import Websocket

import logging
logger = logging.getLogger('websockets')

loop = asyncio.get_event_loop()
ws = Websocket('wss://', loop=loop)
props = loop.run_until_complete(ws.get_dynamic_global_properties())


  • Most of the classes requires async init because during instantiation some API calls has to be performed:
await Amount('10 FOO')
  • Several math operations are not available for graphenecommon.aio.Amount, graphenecommon.aio.Price objects. This includes multiplication, division etc. This limitation is due to unability to define python magic methods (__mul__, __div__, etc) as async coroutines

Concurrent RPC calls

When using async version, you can perform multiple RPC calls from different coroutines concurrently. The library will send requests immediately in non-blocking manner. Incoming responses will be properly matched with queries by using “id” field of json-rpc query.


In asyncio version subscription notifications are not handled in callback-based manner. Instead, they are available in self.notifications queue which is asyncio.Queue.


To enable debugging on RPC level, you can raise loglevel on following loggers (don’t forget to set formatter as well):

log = logging.getLogger("websockets")

log = logging.getLogger("grapheneapi")


Current testsuite uses pre-populated object cache, so it doesn’t cover lots of functionality. Asyncio-specific tests could be run via pytest -v tests/test_*aio*.py