feat: adds support for custom networks

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)

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()

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):

ape_etherscan/query.py

@@ -1,18 +1,43 @@
 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
 
 
 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", ""),
         )

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 = ""