Add new client for Etherscan API v2

safe_eth/eth/clients/__init__.py

@@ -12,6 +12,7 @@
     EtherscanClientException,
     EtherscanRateLimitError,
 )
+from .etherscan_client_v2 import EtherscanClientV2
 from .sourcify_client import (
     SourcifyClient,
     SourcifyClientConfigurationProblem,
@@ -25,6 +26,7 @@
     "ContractMetadata",
     "EnsClient",
     "EtherscanClient",
+    "EtherscanClientV2",
     "EtherscanClientConfigurationProblem",
     "EtherscanClientException",
     "EtherscanRateLimitError",

safe_eth/eth/clients/etherscan_client.py

@@ -319,8 +319,8 @@ def __init__(
         self.http_session.headers = self.HTTP_HEADERS
         self.request_timeout = request_timeout
 
-    def build_url(self, path: str):
-        url = urljoin(self.base_api_url, path)
+    def build_url(self, query: str):
+        url = urljoin(self.base_api_url, f"api?{query}")
         if self.api_key:
             url += f"&apikey={self.api_key}"
         return url
@@ -385,7 +385,7 @@ def get_contract_source_code(self, contract_address: str, retry: bool = True):
         :return:
         """
         url = self.build_url(
-            f"api?module=contract&action=getsourcecode&address={contract_address}"
+            f"module=contract&action=getsourcecode&address={contract_address}"
         )
         response = self._retry_request(url, retry=retry)  # Returns a list
         if response and isinstance(response, list):
@@ -404,7 +404,7 @@ def get_contract_source_code(self, contract_address: str, retry: bool = True):
 
     def get_contract_abi(self, contract_address: str, retry: bool = True):
         url = self.build_url(
-            f"api?module=contract&action=getabi&address={contract_address}"
+            f"module=contract&action=getabi&address={contract_address}"
         )
         result = self._retry_request(url, retry=retry)
         if isinstance(result, dict):

safe_eth/eth/clients/etherscan_client_v2.py

@@ -0,0 +1,56 @@
+import os
+from typing import Any, Dict, List, Optional
+from urllib.parse import urljoin
+
+import requests
+
+from safe_eth.eth import EthereumNetwork
+from safe_eth.eth.clients import EtherscanClient
+
+
+class EtherscanClientV2(EtherscanClient):
+    BASE_API_V2_URL = "https://api.etherscan.io"
+
+    def __init__(
+        self,
+        network: EthereumNetwork,
+        api_key: Optional[str] = None,
+        request_timeout: int = int(
+            os.environ.get("ETHERSCAN_CLIENT_REQUEST_TIMEOUT", 10)
+        ),
+    ):
+        super().__init__(EthereumNetwork.MAINNET, api_key, request_timeout)
+        self.network = network
+        self.base_api_url = self.BASE_API_V2_URL
+
+    def build_url(self, query: str) -> str:
+        url = urljoin(self.base_api_url, f"v2/api?chainid={self.network.value}&{query}")
+        if self.api_key:
+            url += f"&apikey={self.api_key}"
+        return url
+
+    @classmethod
+    def get_supported_networks(cls) -> List[Dict[str, Any]]:
+        """
+        Fetches a list of supported networks by the Etherscan API v2.
+
+        :return: List of supported networks, or empty list if request fails.
+        """
+        url = urljoin(cls.BASE_API_V2_URL, "v2/chainlist")
+        response = requests.get(url)
+        if response.ok:
+            return response.json().get("result", [])
+        return []
+
+    @classmethod
+    def is_supported_network(cls, network: EthereumNetwork) -> bool:
+        """
+        Checks if a given Ethereum network is supported by the Etherscan API v2.
+
+        :param network: The Ethereum network to check.
+        :return: True if the network is supported; False otherwise.
+        """
+        supported_networks = cls.get_supported_networks()
+        return any(
+            item.get("chainid") == str(network.value) for item in supported_networks
+        )

safe_eth/eth/tests/clients/test_etherscan_client_v2.py

@@ -0,0 +1,53 @@
+import os
+
+from django.test import TestCase
+
+import pytest
+
+from ... import EthereumNetwork
+from ...clients import EtherscanClientV2, EtherscanRateLimitError
+from .mocks import sourcify_safe_metadata
+
+
+class TestEtherscanClientV2(TestCase):
+    @classmethod
+    def get_etherscan_api(cls, network: EthereumNetwork):
+        etherscan_api_key_variable_name = "ETHERSCAN_API_KEY"
+        etherscan_api_key = os.environ.get(etherscan_api_key_variable_name)
+        if not etherscan_api_key:
+            pytest.skip(f"{etherscan_api_key_variable_name} needs to be defined")
+
+        return EtherscanClientV2(network, api_key=etherscan_api_key)
+
+    def test_etherscan_get_abi(self):
+        try:
+            etherscan_api = self.get_etherscan_api(EthereumNetwork.MAINNET)
+            safe_master_copy_abi = sourcify_safe_metadata["output"]["abi"]
+            safe_master_copy_address = "0x6851D6fDFAfD08c0295C392436245E5bc78B0185"
+            self.assertEqual(
+                etherscan_api.get_contract_abi(safe_master_copy_address),
+                safe_master_copy_abi,
+            )
+
+            contract_metadata = etherscan_api.get_contract_metadata(
+                safe_master_copy_address
+            )
+            self.assertEqual(contract_metadata.name, "GnosisSafe")
+            self.assertEqual(contract_metadata.abi, safe_master_copy_abi)
+
+            random_address = "0xaE32496491b53841efb51829d6f886387708F99a"
+            self.assertIsNone(etherscan_api.get_contract_abi(random_address))
+            self.assertIsNone(etherscan_api.get_contract_metadata(random_address))
+        except EtherscanRateLimitError:
+            self.skipTest("Etherscan rate limit reached")
+
+    def test_is_supported_network(self):
+        try:
+            self.assertTrue(
+                EtherscanClientV2.is_supported_network(EthereumNetwork.GNOSIS)
+            )
+            self.assertFalse(
+                EtherscanClientV2.is_supported_network(EthereumNetwork.UNKNOWN)
+            )
+        except EtherscanRateLimitError:
+            self.skipTest("Etherscan rate limit reached")