diff --git a/boa/explorer.py b/boa/explorer.py index 8a9cc4d7..edef0ab7 100644 --- a/boa/explorer.py +++ b/boa/explorer.py @@ -24,38 +24,63 @@ @dataclass -class ExplorerSettings: +class BlockExplorer: api_key: Optional[str] = os.environ.get("ETHERSCAN_API_KEY") uri: str = os.environ.get("ETHERSCAN_URI", "https://api.etherscan.io/api") + num_retries: int = 10 + backoff_ms: int = 400 + + def fetch(self, **params) -> dict: + """ + Fetch data from Etherscan API. + Offers a simple caching mechanism to avoid redundant queries. + Retries if rate limit is reached. + :param num_retries: Number of retries + :param backoff_ms: Backoff in milliseconds + :param params: Additional query parameters + :return: JSON response + """ + params = {**params, "apiKey": self.api_key} + for i in range(self.num_retries): + res = SESSION.get(self.uri, params=params) + res.raise_for_status() + data = res.json() + if not _is_rate_limited(data): + break + backoff_factor = 1.1**i # 1.1**10 ~= 2.59 + time.sleep(backoff_factor * self.backoff_ms / 1000) + + if not _is_success_response(data): + raise ValueError(f"Failed to retrieve data from API: {data}") + + return data + + +etherscan = BlockExplorer() def _fetch_etherscan( - settings: ExplorerSettings, num_retries=10, backoff_ms=400, **params + uri: Optional[str], + api_key: Optional[str] = None, + num_retries=10, + backoff_ms=400, + **params, ) -> dict: """ Fetch data from Etherscan API. Offers a simple caching mechanism to avoid redundant queries. Retries if rate limit is reached. - :param settings: Etherscan settings + :param uri: Etherscan API URI + :param api_key: Etherscan API key :param num_retries: Number of retries :param backoff_ms: Backoff in milliseconds :param params: Additional query parameters :return: JSON response """ - params = {**params, "apiKey": settings.api_key} - for i in range(num_retries): - res = SESSION.get(settings.uri, params=params) - res.raise_for_status() - data = res.json() - if not _is_rate_limited(data): - break - backoff_factor = 1.1**i # 1.1**10 ~= 2.59 - time.sleep(backoff_factor * backoff_ms / 1000) - - if not _is_success_response(data): - raise ValueError(f"Failed to retrieve data from API: {data}") - - return data + uri = uri or etherscan.uri + api_key = api_key or etherscan.api_key + explorer = BlockExplorer(api_key, uri, num_retries, backoff_ms) + return explorer.fetch(**params) def _is_success_response(data: dict) -> bool: @@ -74,24 +99,24 @@ def _is_rate_limited(data: dict) -> bool: return "rate limit" in data.get("result", "") and data.get("status") == "0" -def fetch_abi_from_etherscan( - address: str, settings: ExplorerSettings = ExplorerSettings() -): +def fetch_abi_from_etherscan(address: str, uri: str = None, api_key: str = None): # resolve implementation address if `address` is a proxy contract - address = _resolve_implementation_address(address, settings) + address = _resolve_implementation_address(address, uri, api_key) # fetch ABI of `address` params = dict(module="contract", action="getabi", address=address) - data = _fetch_etherscan(settings, **params) + data = _fetch_etherscan(uri, api_key, **params) return json.loads(data["result"].strip()) # fetch the address of a contract; resolves at most one layer of indirection # if the address is a proxy contract. -def _resolve_implementation_address(address: str, settings: ExplorerSettings): +def _resolve_implementation_address( + address: str, uri: Optional[str], api_key: Optional[str] +): params = dict(module="contract", action="getsourcecode", address=address) - data = _fetch_etherscan(settings, **params) + data = _fetch_etherscan(uri, api_key, **params) source_data = data["result"][0] # check if the contract is a proxy diff --git a/boa/interpret.py b/boa/interpret.py index 8ee512e4..18e62de6 100644 --- a/boa/interpret.py +++ b/boa/interpret.py @@ -28,7 +28,7 @@ VyperDeployer, ) from boa.environment import Env -from boa.explorer import ExplorerSettings, fetch_abi_from_etherscan +from boa.explorer import fetch_abi_from_etherscan from boa.rpc import json from boa.util.abi import Address from boa.util.disk_cache import DiskCache @@ -41,7 +41,6 @@ _disk_cache = None _search_path = None -explorer_settings = ExplorerSettings() def set_search_path(path: list[str]): @@ -254,12 +253,11 @@ def _compile(): return _disk_cache.caching_lookup(cache_key, _compile) -def from_etherscan(address: Any, name=None, uri=None, api_key=None): +def from_etherscan( + address: Any, name: str = None, uri: str = None, api_key: str = None +): addr = Address(address) - api_key = api_key or explorer_settings.api_key - uri = uri or explorer_settings.uri - settings = ExplorerSettings(api_key, uri) - abi = fetch_abi_from_etherscan(addr, settings) + abi = fetch_abi_from_etherscan(addr, uri, api_key) return ABIContractFactory.from_abi_dict(abi, name=name).at(addr) diff --git a/tests/integration/fork/test_from_etherscan.py b/tests/integration/fork/test_from_etherscan.py index 662dac6e..a380be98 100644 --- a/tests/integration/fork/test_from_etherscan.py +++ b/tests/integration/fork/test_from_etherscan.py @@ -16,7 +16,7 @@ @pytest.fixture(scope="module", autouse=True) def api_key(): - boa.interpret.explorer_settings.api_key = os.environ["ETHERSCAN_API_KEY"] + boa.explorer.etherscan.api_key = os.environ["ETHERSCAN_API_KEY"] @pytest.fixture(scope="module")