Using AsyncWeb3 with trio in python

test_particle_async_ws.py

import os
import dotenv
import trio
import trio_asyncio
from web3 import AsyncWeb3
from web3 import WebsocketProviderV2

dotenv.load_dotenv()


class ParticleConfigEndpoint:
    def __init__(self, chain_id: int = 1):
        self.project_id = os.getenv('PROJECT_ID')
        self.project_server_key = os.getenv('PROJECT_SERVER_KEY')
        self.chain_id = chain_id
        self.endpoint = f"wss://rpc.particle.network/evm-chain"
        self.endpoint_uri = self.endpoint + f'?chainId={chain_id}&projectUuid={self.project_id}&projectKey={self.project_server_key}'
        self.request_timeout = 60

# Get the endpoint URL for the chain that we want to connect to.
ep = ParticleConfigEndpoint(1).endpoint_uri


async def ws_v2_alternate_init_example_1():
    # compatability hack because AsyncWeb3 needs an event_loop and trio does not provide one 
    async with trio_asyncio.open_loop():
        w3: AsyncWeb3 = await trio_asyncio.aio_as_trio(AsyncWeb3.persistent_websocket(WebsocketProviderV2(ep)))
        print(await trio_asyncio.aio_as_trio(w3.eth.get_block('latest', full_transactions=True)))
        # we cannot use context managers, so we need to manually disconnect the websocket when we are done. 
        await trio_asyncio.aio_as_trio(w3.provider.disconnect())

if __name__ == "__main__":
    # the best part is that we can still directly use trio.run for MultiCore, asynchronous, GIL bypassing awesomeness! 
    trio.run(ws_v2_alternate_init_example_1)