Using web3py with asyncio

web3py_aio.py

"""
Inspired by this approach:

- https://twitter.com/zzzeek/status/1279069782533386247
- https://gist.github.com/zzzeek/2a8d94b03e46b8676a063a32f78140f1
- https://gist.github.com/zzzeek/33943060f7a08cf9e82bf8df1f0f75de, https://gist.github.com/zzzeek/4e89ce6226826e7a8df13e1b573ad354#file-asyncio_plus_greenlet-py-L28

Essentially we use greenlets to make the IO layer async and the top layer uses async/await, but everything in
between can be regular sync code.

Other gists and libs along those lines:
- https://gist.github.com/zzzeek/769b684d4fc8dfec9d4ebc6e4bb93076
- https://gist.github.com/snaury/202bf4f22c41ca34e56297bae5f33fef
- https://github.com/oremanj/greenback

Two parts to use it:

- Wrap any low-level async IO function in `green_await()`
- In your high-level async code, call your sync-cde via `green_spawn`.

So:

    provider = AIOHTTPProvider(current_app.config['ETHEREUM_API_URL'])    
    web3 = Web3(provider=provider)
    
    def do_async_web3():
        print(web3.eth.block_number)
    green_spawn(do_async_web3)
"""
import sys
from typing import Any

import aiohttp
from sqlalchemy.util import await_only
from web3 import HTTPProvider
from web3.types import RPCEndpoint, RPCResponse
from eth_typing import URI

import greenlet


# From: https://gist.github.com/snaury/202bf4f22c41ca34e56297bae5f33fef
class AsyncIoGreenlet(greenlet.greenlet):
    def __init__(self, driver, fn):
        greenlet.greenlet.__init__(self, fn, driver)
        self.driver = driver


async def green_spawn(fn, *args, **kwargs):
    context = AsyncIoGreenlet(greenlet.getcurrent(), fn)

    result = context.switch(*args, **kwargs)

    while context:
        try:
            value = await result
        except:
            result = context.throw(*sys.exc_info())
        else:
            result = context.switch(value)

    return result


def green_await(awaitable):
    current = greenlet.getcurrent()
    if not isinstance(current, AsyncIoGreenlet):
        raise TypeError('Cannot use green_await outside of green_spawn target')
    return current.driver.switch(awaitable)


class AIOHTTPProvider(HTTPProvider):

    def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
        self.logger.debug("Making request HTTP. URI: %s, Method: %s",
                          self.endpoint_uri, method)
        request_data = self.encode_rpc_request(method, params)
        raw_response = await_only(make_post_request(
            self.endpoint_uri,
            request_data,
            **self.get_request_kwargs()
        ))
        response = self.decode_rpc_response(raw_response)
        self.logger.debug("Getting response HTTP. URI: %s, "
                          "Method: %s, Response: %s",
                          self.endpoint_uri, method, response)
        return response


async def make_post_request(endpoint_uri: URI, data: bytes, *args: Any, **kwargs: Any) -> bytes:
    kwargs.setdefault('timeout', 10)
    async with aiohttp.ClientSession() as client:
        response = await client.post(endpoint_uri, data=data, *args, **kwargs)  # type: ignore
        response.raise_for_status()
        return await response.content.read()

How has this not gotten more attention, this is awesome, great work dude!