async-etherscan-python

async-etherscan-python.py

"""
Useage:
    api_key = ""
    api = Etherscan(api_key)

    async def get_block_number_by_timestamp(timestamp):
        return await api.get_block_number_by_timestamp(timestamp, closest="before") 
"""
import json
from importlib import resources
import etherscan
from etherscan import configs
from etherscan.enums.fields_enum import FieldsEnum as fields
from etherscan.utils.parsing import ResponseParser as parser
import aiohttp


class Etherscan:
    def __new__(cls, api_key: str, net: str = "MAIN"):
        with resources.path(configs, f"{net.upper()}-stable.json") as path:
            config_path = str(path)
        return cls.from_config(api_key=api_key, config_path=config_path, net=net)

    @staticmethod
    def __load_config(config_path: str) -> dict:
        with open(config_path, "r") as f:
            return json.load(f)

    @staticmethod
    def __run(func, api_key: str, net: str):
        async def wrapper(*args, **kwargs):
            url = (
                f"{fields.PREFIX.format(net.lower()).replace('-main','')}"
                f"{func(*args, **kwargs)}"
                f"{fields.API_KEY}"
                f"{api_key}"
            )
            async with aiohttp.ClientSession(
                headers={"User-Agent": ""}, raise_for_status=True
            ) as client:
                response = await client.get(url)
                content = await response.json()
                result = content["result"]
                if "status" in content.keys():
                    status = bool(int(content["status"]))
                    message = content["message"]
                    assert status, f"{result} -- {message}"
                else:
                    pass
                return result

        return wrapper

    @classmethod
    def from_config(cls, api_key: str, config_path: str, net: str):
        config = cls.__load_config(config_path)
        for func, v in config.items():
            if not func.startswith("_"):  # disabled if _
                attr = getattr(getattr(etherscan, v["module"]), func)
                setattr(cls, func, cls.__run(attr, api_key, net))
        return cls