Multichain Web3

multichain_w3.py

import asyncio
import logging
import os

from web3.middleware import geth_poa_middleware

import utils.termstyle
import web3
cp = utils.termstyle.ColorPrinter(0)


class MultiChainWeb3:
    def __init__(self, _networks: list[str], logger: logging.Logger):
        self.networks = _networks
        self.logger = logger
        self.w3_dict = {}
        self.setup()

    def setup(self):
        for network in self.networks:
            self.conf_w3(network)

    def conf_w3(self, network: str) -> (web3.Web3, str):
        endpoint = os.environ.get('%s_ws_endpoint' % network)
        if not endpoint:
            cp.err('Unable to configure web3 for %s' % network)

        conn = web3.Web3(web3.WebsocketProvider(endpoint))
        if conn.is_connected():
            cp.good('%s connected' % network)
            self.w3_dict.update({network: {'connection': conn, 'endpoint': endpoint}})
        else:
            cp.err('Error: %s could not connect' % network)
        if conn.eth.chain_id in [10, 137, 56]:
            conn.middleware_onion.inject(geth_poa_middleware, layer=0)
        return conn, endpoint

    def get_w3(self, network):
        cp.debug('DEBUG: get w3 %s' % network)
        return self.w3_dict.get(network).get('connection')

    async def reconnect_w3(self, network, delay: int = 3):
        await asyncio.sleep(delay)
        try:
            conn, endpoint = self.conf_w3(network)
        except Exception as err:
            self.logger.error('Error in reconnect_w3: "%s" , sleep: %s, Trying again.' % (err, delay))
            return await self.reconnect_w3(network, delay*2)
        else:
            self.w3_dict.update({network: {'connection': conn, 'endpoint': endpoint}})