From bf4c0a5a57531ccf5511e822c3f3b47ccdd147db Mon Sep 17 00:00:00 2001 From: Mike Shultz Date: Fri, 12 Jan 2024 14:30:37 -0700 Subject: [PATCH] feat: adds support for custom networks --- ape_etherscan/client.py | 61 +++++++++++++++++++++++++++------------ ape_etherscan/config.py | 22 ++++++++++++++ ape_etherscan/explorer.py | 40 ++++++++++++++++++------- ape_etherscan/query.py | 29 +++++++++++++++++-- ape_etherscan/types.py | 8 +++++ 5 files changed, 129 insertions(+), 31 deletions(-) diff --git a/ape_etherscan/client.py b/ape_etherscan/client.py index e774d4c..5f005be 100644 --- a/ape_etherscan/client.py +++ b/ape_etherscan/client.py @@ -5,21 +5,41 @@ from io import StringIO from typing import Dict, Iterator, List, Optional +from ape.api import PluginConfig from ape.logging import logger from ape.utils import USER_AGENT, ManagerAccessMixin from requests import Session from yarl import URL +from ape_etherscan.config import EtherscanConfig from ape_etherscan.exceptions import ( UnhandledResultError, UnsupportedEcosystemError, UnsupportedNetworkError, ) -from ape_etherscan.types import ContractCreationResponse, EtherscanResponse, SourceCodeResponse +from ape_etherscan.types import ( + ContractCreationResponse, + EtherscanInstance, + EtherscanResponse, + SourceCodeResponse, +) from ape_etherscan.utils import API_KEY_ENV_KEY_MAP -def get_etherscan_uri(ecosystem_name: str, network_name: str): +def get_network_config( + etherscan_config: EtherscanConfig, ecosystem_name: str, network_name: str +) -> Optional[PluginConfig]: + if ecosystem_name in etherscan_config: + return etherscan_config[ecosystem_name].get(network_name) + return None + + +def get_etherscan_uri(etherscan_config: EtherscanConfig, ecosystem_name: str, network_name: str): + # Look for explicitly configured Etherscan config + network_conf = get_network_config(etherscan_config, ecosystem_name, network_name) + if network_conf and hasattr(network_conf, "uri"): + return network_conf.uri + if ecosystem_name == "ethereum": return ( f"https://{network_name}.etherscan.io" @@ -80,7 +100,14 @@ def get_etherscan_uri(ecosystem_name: str, network_name: str): raise UnsupportedEcosystemError(ecosystem_name) -def get_etherscan_api_uri(ecosystem_name: str, network_name: str): +def get_etherscan_api_uri( + etherscan_config: EtherscanConfig, ecosystem_name: str, network_name: str +): + # Look for explicitly configured Etherscan config + network_conf = get_network_config(etherscan_config, ecosystem_name, network_name) + if network_conf and hasattr(network_conf, "api_uri"): + return network_conf.uri + if ecosystem_name == "ethereum": return ( f"https://api-{network_name}.etherscan.io/api" @@ -149,15 +176,14 @@ class _APIClient(ManagerAccessMixin): DEFAULT_HEADERS = {"User-Agent": USER_AGENT} session = Session() - def __init__(self, ecosystem_name: str, network_name: str, module_name: str): - self._ecosystem_name = ecosystem_name - self._network_name = network_name + def __init__(self, instance: EtherscanInstance, module_name: str): + self._instance = instance self._module_name = module_name self._last_call = 0.0 @property def base_uri(self) -> str: - return get_etherscan_api_uri(self._ecosystem_name, self._network_name) + return self._instance.api_uri @property def base_params(self) -> Dict: @@ -245,10 +271,10 @@ def _request( break - return EtherscanResponse(response, self._ecosystem_name, raise_on_exceptions) + return EtherscanResponse(response, self._instance.ecosystem_name, raise_on_exceptions) def __authorize(self, params_or_data: Optional[Dict] = None) -> Optional[Dict]: - env_var_key = API_KEY_ENV_KEY_MAP.get(self._ecosystem_name) + env_var_key = API_KEY_ENV_KEY_MAP.get(self._instance.ecosystem_name) if not env_var_key: return params_or_data @@ -262,9 +288,9 @@ def __authorize(self, params_or_data: Optional[Dict] = None) -> Optional[Dict]: class ContractClient(_APIClient): - def __init__(self, ecosystem_name: str, network_name: str, address: str): + def __init__(self, instance: EtherscanInstance, address: str): self._address = address - super().__init__(ecosystem_name, network_name, "contract") + super().__init__(instance, "contract") def get_source_code(self) -> SourceCodeResponse: params = { @@ -357,9 +383,9 @@ def get_creation_data(self) -> List[ContractCreationResponse]: class AccountClient(_APIClient): - def __init__(self, ecosystem_name: str, network_name: str, address: str): + def __init__(self, instance: EtherscanInstance, address: str): self._address = address - super().__init__(ecosystem_name, network_name, "account") + super().__init__(instance, "account") def get_all_normal_transactions( self, @@ -408,12 +434,11 @@ def _get_page_of_normal_transactions( class ClientFactory: - def __init__(self, ecosystem_name: str, network_name: str): - self._ecosystem_name = ecosystem_name - self._network_name = network_name + def __init__(self, instance: EtherscanInstance): + self._instance = instance def get_contract_client(self, contract_address: str) -> ContractClient: - return ContractClient(self._ecosystem_name, self._network_name, contract_address) + return ContractClient(self._instance, contract_address) def get_account_client(self, account_address: str) -> AccountClient: - return AccountClient(self._ecosystem_name, self._network_name, account_address) + return AccountClient(self._instance, account_address) diff --git a/ape_etherscan/config.py b/ape_etherscan/config.py index 14dbd23..e112b35 100644 --- a/ape_etherscan/config.py +++ b/ape_etherscan/config.py @@ -1,12 +1,34 @@ +from typing import Optional + from ape.api.config import PluginConfig +from pydantic import AnyHttpUrl, model_validator +from pydantic_settings import SettingsConfigDict + + +class NetworkConfig(PluginConfig): + uri: Optional[AnyHttpUrl] = None + api_uri: Optional[AnyHttpUrl] = None class EcosystemConfig(PluginConfig): + model_config = SettingsConfigDict(extra="allow") + rate_limit: int = 5 # Requests per second retries: int = 5 # Number of retries before giving up + @model_validator(mode="after") + def verify_extras(self) -> "EcosystemConfig": + if self.__pydantic_extra__: + for aname in self.__pydantic_extra__.keys(): + self.__pydantic_extra__[aname] = NetworkConfig.parse_obj( + self.__pydantic_extra__[aname] + ) + return self + class EtherscanConfig(PluginConfig): + model_config = SettingsConfigDict(extra="allow") + ethereum: EcosystemConfig = EcosystemConfig() arbitrum: EcosystemConfig = EcosystemConfig() fantom: EcosystemConfig = EcosystemConfig() diff --git a/ape_etherscan/explorer.py b/ape_etherscan/explorer.py index 000b032..ab442ea 100644 --- a/ape_etherscan/explorer.py +++ b/ape_etherscan/explorer.py @@ -2,32 +2,50 @@ from json.decoder import JSONDecodeError from typing import Optional -from ape.api import ExplorerAPI +from ape.api import ExplorerAPI, PluginConfig from ape.contracts import ContractInstance from ape.exceptions import ProviderNotConnectedError from ape.logging import logger from ape.types import AddressType, ContractType -from ape_etherscan.client import ClientFactory, get_etherscan_uri +from ape_etherscan.client import ClientFactory, get_etherscan_api_uri, get_etherscan_uri +from ape_etherscan.types import EtherscanInstance from ape_etherscan.verify import SourceVerifier class Etherscan(ExplorerAPI): - def get_address_url(self, address: str) -> str: - etherscan_uri = get_etherscan_uri( - self.network.ecosystem.name, self.network.name.replace("-fork", "") + @property + def _config(self) -> PluginConfig: + return self.config_manager.get_config("etherscan") + + @property + def etherscan_uri(self): + return get_etherscan_uri( + self._config, self.network.ecosystem.name, self.network.name.replace("-fork", "") ) - return f"{etherscan_uri}/address/{address}" - def get_transaction_url(self, transaction_hash: str) -> str: - etherscan_uri = get_etherscan_uri( - self.network.ecosystem.name, self.network.name.replace("-fork", "") + @property + def etherscan_api_uri(self): + return get_etherscan_api_uri( + self._config, self.network.ecosystem.name, self.network.name.replace("-fork", "") ) - return f"{etherscan_uri}/tx/{transaction_hash}" + + def get_address_url(self, address: str) -> str: + return f"{self.etherscan_uri}/address/{address}" + + def get_transaction_url(self, transaction_hash: str) -> str: + return f"{self.etherscan_uri}/tx/{transaction_hash}" @property def _client_factory(self) -> ClientFactory: - return ClientFactory(self.network.ecosystem.name, self.network.name.replace("-fork", "")) + return ClientFactory( + EtherscanInstance( + ecosystem_name=self.network.ecosystem.name, + network_name=self.network.name.replace("-fork", ""), + uri=self.etherscan_uri, + api_uri=self.etherscan_api_uri, + ) + ) def get_contract_type(self, address: AddressType) -> Optional[ContractType]: if not self.conversion_manager.is_type(address, AddressType): diff --git a/ape_etherscan/query.py b/ape_etherscan/query.py index d4b4384..1a76165 100644 --- a/ape_etherscan/query.py +++ b/ape_etherscan/query.py @@ -1,11 +1,12 @@ from typing import Iterator, Optional -from ape.api import QueryAPI, QueryType, ReceiptAPI +from ape.api import PluginConfig, QueryAPI, QueryType, ReceiptAPI from ape.api.query import AccountTransactionQuery, ContractCreationQuery from ape.exceptions import QueryEngineError from ape.utils import singledispatchmethod -from ape_etherscan.client import ClientFactory +from ape_etherscan.client import ClientFactory, get_etherscan_api_uri, get_etherscan_uri +from ape_etherscan.types import EtherscanInstance from ape_etherscan.utils import NETWORKS @@ -13,6 +14,30 @@ class EtherscanQueryEngine(QueryAPI): @property def _client_factory(self) -> ClientFactory: return ClientFactory( + EtherscanInstance( + ecosystem_name=self.provider.network.ecosystem.name, + network_name=self.provider.network.name.replace("-fork", ""), + uri=self.etherscan_uri, + api_uri=self.etherscan_api_uri, + ) + ) + + @property + def _config(self) -> PluginConfig: + return self.config_manager.get_config("etherscan") + + @property + def etherscan_uri(self): + return get_etherscan_uri( + self._config, + self.provider.network.ecosystem.name, + self.provider.network.name.replace("-fork", ""), + ) + + @property + def etherscan_api_uri(self): + return get_etherscan_api_uri( + self._config, self.provider.network.ecosystem.name, self.provider.network.name.replace("-fork", ""), ) diff --git a/ape_etherscan/types.py b/ape_etherscan/types.py index 82165ea..58533df 100644 --- a/ape_etherscan/types.py +++ b/ape_etherscan/types.py @@ -7,6 +7,14 @@ from ape_etherscan.exceptions import EtherscanResponseError, get_request_error +@dataclass +class EtherscanInstance: + ecosystem_name: str + network_name: str # normalized (e.g. no -fork) + uri: str + api_uri: str + + @dataclass class SourceCodeResponse: abi: str = ""