10793540350

Committed 10 Sep 2024 01:31PM UTC coverage: 93.551% (-0.3%) from 93.892%

Build # 10793540350

Build Type

push

github

Committed by

falvaradorodriguez

Commit Message

Fix cowswap test

Run Details

8777 of 9382 relevant lines covered (93.55%)

3.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.71

/safe_eth/eth/clients/etherscan_client.py

import json
import os
import time
from typing import Any, Dict, List, MutableMapping, Optional, Union
from urllib.parse import urljoin

from ...util.http import prepare_http_session
from .. import EthereumNetwork
from .contract_metadata import ContractMetadata


class EtherscanClientException(Exception):
    pass


class EtherscanClientConfigurationProblem(Exception):
    pass


class EtherscanRateLimitError(EtherscanClientException):
    pass


class EtherscanClient:
    NETWORK_WITH_URL = {
        EthereumNetwork.MAINNET: "https://etherscan.io",
        EthereumNetwork.RINKEBY: "https://rinkeby.etherscan.io",
        EthereumNetwork.ROPSTEN: "https://ropsten.etherscan.io",
        EthereumNetwork.GOERLI: "https://goerli.etherscan.io",
        EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://bscscan.com",
        EthereumNetwork.POLYGON: "https://polygonscan.com",
        EthereumNetwork.POLYGON_ZKEVM: "https://zkevm.polygonscan.com",
        EthereumNetwork.OPTIMISM: "https://optimistic.etherscan.io",
        EthereumNetwork.ARBITRUM_ONE: "https://arbiscan.io",
        EthereumNetwork.ARBITRUM_NOVA: "https://nova.arbiscan.io",
        EthereumNetwork.ARBITRUM_GOERLI: "https://goerli.arbiscan.io",
        EthereumNetwork.AVALANCHE_C_CHAIN: "https://snowtrace.io",
        EthereumNetwork.GNOSIS: "https://gnosisscan.io",
        EthereumNetwork.MOONBEAM: "https://moonbeam.moonscan.io",
        EthereumNetwork.MOONRIVER: "https://moonriver.moonscan.io",
        EthereumNetwork.MOONBASE_ALPHA: "https://moonbase.moonscan.io",
        EthereumNetwork.CRONOS_MAINNET: "https://cronoscan.com",
        EthereumNetwork.CRONOS_TESTNET: "https://testnet.cronoscan.com",
        EthereumNetwork.CELO_MAINNET: "https://celoscan.io",
        EthereumNetwork.BASE_GOERLI_TESTNET: "https://goerli.basescan.org",
        EthereumNetwork.NEON_EVM_DEVNET: "https://devnet.neonscan.org",
        EthereumNetwork.NEON_EVM_MAINNET: "https://neonscan.org",
        EthereumNetwork.SEPOLIA: "https://sepolia.etherscan.io",
        EthereumNetwork.ZKSYNC_MAINNET: "https://explorer.zksync.io/",
        EthereumNetwork.FANTOM_OPERA: "https://ftmscan.com",
        EthereumNetwork.FANTOM_TESTNET: "https://testnet.ftmscan.com/",
        EthereumNetwork.LINEA: "https://lineascan.build",
        EthereumNetwork.LINEA_GOERLI: "https://goerli.lineascan.build",
        EthereumNetwork.MANTLE: "https://explorer.mantle.xyz",
        EthereumNetwork.MANTLE_TESTNET: "https://explorer.testnet.mantle.xyz",
        EthereumNetwork.JAPAN_OPEN_CHAIN_MAINNET: "https://mainnet.japanopenchain.org",
        EthereumNetwork.JAPAN_OPEN_CHAIN_TESTNET: "https://explorer.testnet.japanopenchain.org",
        EthereumNetwork.SCROLL_SEPOLIA_TESTNET: "https://sepolia.scrollscan.dev",
        EthereumNetwork.SCROLL: "https://scrollscan.com",
        EthereumNetwork.KROMA: "https://kromascan.com",
        EthereumNetwork.KROMA_SEPOLIA: "https://sepolia.kromascan.com",
        EthereumNetwork.BLAST_SEPOLIA_TESTNET: "https://sepolia.blastscan.io",
        EthereumNetwork.FRAXTAL: "https://fraxscan.com",
        EthereumNetwork.BASE: "https://api.basescan.org/",
        EthereumNetwork.BLAST: "https://blastscan.io",
        EthereumNetwork.TAIKO_MAINNET: "https://taikoscan.io",
        EthereumNetwork.BASE_SEPOLIA_TESTNET: "https://sepolia.basescan.org",
        EthereumNetwork.HOLESKY: "https://holesky.etherscan.io",
        EthereumNetwork.LINEA_SEPOLIA: "https://sepolia.lineascan.build",
    }

    NETWORK_WITH_API_URL = {
        EthereumNetwork.MAINNET: "https://api.etherscan.io",
        EthereumNetwork.RINKEBY: "https://api-rinkeby.etherscan.io",
        EthereumNetwork.ROPSTEN: "https://api-ropsten.etherscan.io",
        EthereumNetwork.GOERLI: "https://api-goerli.etherscan.io",
        EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://api.bscscan.com",
        EthereumNetwork.POLYGON: "https://api.polygonscan.com",
        EthereumNetwork.POLYGON_ZKEVM: "https://api-zkevm.polygonscan.com",
        EthereumNetwork.OPTIMISM: "https://api-optimistic.etherscan.io",
        EthereumNetwork.ARBITRUM_ONE: "https://api.arbiscan.io",
        EthereumNetwork.ARBITRUM_NOVA: "https://api-nova.arbiscan.io",
        EthereumNetwork.ARBITRUM_GOERLI: "https://api-goerli.arbiscan.io",
        EthereumNetwork.ARBITRUM_SEPOLIA: "https://api-sepolia.arbiscan.io",
        EthereumNetwork.AVALANCHE_C_CHAIN: "https://api.snowtrace.io",
        EthereumNetwork.GNOSIS: "https://api.gnosisscan.io",
        EthereumNetwork.MOONBEAM: "https://api-moonbeam.moonscan.io",
        EthereumNetwork.MOONRIVER: "https://api-moonriver.moonscan.io",
        EthereumNetwork.MOONBASE_ALPHA: "https://api-moonbase.moonscan.io",
        EthereumNetwork.CRONOS_MAINNET: "https://api.cronoscan.com",
        EthereumNetwork.CRONOS_TESTNET: "https://api-testnet.cronoscan.com",
        EthereumNetwork.CELO_MAINNET: "https://api.celoscan.io",
        EthereumNetwork.BASE_GOERLI_TESTNET: "https://api-goerli.basescan.org",
        EthereumNetwork.NEON_EVM_DEVNET: "https://devnet-api.neonscan.org",
        EthereumNetwork.NEON_EVM_MAINNET: "https://api.neonscan.org",
        EthereumNetwork.SEPOLIA: "https://api-sepolia.etherscan.io",
        EthereumNetwork.ZKSYNC_MAINNET: "https://block-explorer-api.mainnet.zksync.io/",
        EthereumNetwork.FANTOM_OPERA: "https://api.ftmscan.com",
        EthereumNetwork.FANTOM_TESTNET: "https://api-testnet.ftmscan.com",
        EthereumNetwork.LINEA: "https://api.lineascan.build",
        EthereumNetwork.LINEA_GOERLI: "https://api-testnet.lineascan.build",
        EthereumNetwork.MANTLE: "https://explorer.mantle.xyz",
        EthereumNetwork.MANTLE_TESTNET: "https://explorer.testnet.mantle.xyz",
        EthereumNetwork.JAPAN_OPEN_CHAIN_MAINNET: "https://mainnet.japanopenchain.org/api",
        EthereumNetwork.JAPAN_OPEN_CHAIN_TESTNET: "https://explorer.testnet.japanopenchain.org/api",
        EthereumNetwork.SCROLL_SEPOLIA_TESTNET: "https://api-sepolia.scrollscan.dev",
        EthereumNetwork.SCROLL: "https://api.scrollscan.com",
        EthereumNetwork.KROMA: "https://api.kromascan.com",
        EthereumNetwork.KROMA_SEPOLIA: "https://api-sepolia.kromascan.com",
        EthereumNetwork.BLAST_SEPOLIA_TESTNET: "https://api-sepolia.blastscan.io",
        EthereumNetwork.FRAXTAL: "https://api.fraxscan.com",
        EthereumNetwork.BASE: "https://api.basescan.org",
        EthereumNetwork.BLAST: "https://api.blastscan.io",
        EthereumNetwork.TAIKO_MAINNET: "https://api.taikoscan.io",
        EthereumNetwork.BASE_SEPOLIA_TESTNET: "https://api-sepolia.basescan.org/api",
        EthereumNetwork.HOLESKY: "https://api-holesky.etherscan.io",
        EthereumNetwork.LINEA_SEPOLIA: "https://api-sepolia.lineascan.build",
    }
    HTTP_HEADERS: MutableMapping[str, Union[str, bytes]] = {
        "User-Agent": "curl/7.77.0",
    }

    def __init__(
        self,
        network: EthereumNetwork,
        api_key: Optional[str] = None,
        request_timeout: int = int(
            os.environ.get("ETHERSCAN_CLIENT_REQUEST_TIMEOUT", 10)
        ),
    ):
        self.network = network
        self.api_key = api_key
        self.base_url = self.NETWORK_WITH_URL.get(network, "")
        self.base_api_url = self.NETWORK_WITH_API_URL.get(network, "")
        if not self.base_api_url:
            raise EtherscanClientConfigurationProblem(
                f"Network {network.name} - {network.value} not supported"
            )
        self.http_session = prepare_http_session(10, 100)
        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)
        if self.api_key:
            url += f"&apikey={self.api_key}"
        return url

    def _do_request(self, url: str) -> Optional[Union[Dict[str, Any], List[Any], str]]:
        response = self.http_session.get(url, timeout=self.request_timeout)

        if response.ok:
            response_json = response.json()
            result = response_json["result"]
            if "Max rate limit reached" in result:
                # Max rate limit reached, please use API Key for higher rate limit
                raise EtherscanRateLimitError
            if response_json["status"] == "1":
                return result
        return None

    def _retry_request(
        self, url: str, retry: bool = True
    ) -> Optional[Union[Dict[str, Any], List[Any], str]]:
        for _ in range(3):
            try:
                return self._do_request(url)
            except EtherscanRateLimitError as exc:
                if not retry:
                    raise exc
                else:
                    time.sleep(5)
        return None

    def get_contract_metadata(
        self, contract_address: str, retry: bool = True
    ) -> Optional[ContractMetadata]:
        contract_source_code = self.get_contract_source_code(
            contract_address, retry=retry
        )
        if contract_source_code:
            contract_name = contract_source_code["ContractName"]
            contract_abi = contract_source_code["ABI"]
            if contract_abi:
                return ContractMetadata(contract_name, contract_abi, False)
        return None

    def get_contract_source_code(self, contract_address: str, retry: bool = True):
        """
        Get source code for a contract. Source code query also returns:

            - ContractName: "",
            - CompilerVersion: "",
            - OptimizationUsed: "",
            - Runs: "",
            - ConstructorArguments: ""
            - EVMVersion: "Default",
            - Library: "",
            - LicenseType: "",
            - Proxy: "0",
            - Implementation: "",
            - SwarmSource: ""

        :param contract_address:
        :param retry: if ``True``, try again if there's Rate Limit Error
        :return:
        """
        url = self.build_url(
            f"api?module=contract&action=getsourcecode&address={contract_address}"
        )
        response = self._retry_request(url, retry=retry)  # Returns a list
        if response and isinstance(response, list):
            result = response[0]
            abi_str = result.get("ABI")

            if isinstance(abi_str, str) and abi_str.startswith("["):
                try:
                    result["ABI"] = json.loads(abi_str)
                except json.JSONDecodeError:
                    result["ABI"] = None  # Handle the case where JSON decoding fails
            else:
                result["ABI"] = None

            return result

    def get_contract_abi(self, contract_address: str, retry: bool = True):
        url = self.build_url(
            f"api?module=contract&action=getabi&address={contract_address}"
        )
        result = self._retry_request(url, retry=retry)
        if isinstance(result, dict):
            return result
        elif isinstance(result, str):
            try:
                return json.loads(result)
            except json.JSONDecodeError:
                pass
        return None

1	import json	4✔
2	import os	4✔
3	import time	4✔
4	from typing import Any, Dict, List, MutableMapping, Optional, Union	4✔
5	from urllib.parse import urljoin	4✔
6
7	from ...util.http import prepare_http_session	4✔
8	from .. import EthereumNetwork	4✔
9	from .contract_metadata import ContractMetadata	4✔
10
11
12	class EtherscanClientException(Exception):	4✔
13	pass	4✔
14
15
16	class EtherscanClientConfigurationProblem(Exception):	4✔
17	pass	4✔
18
19
20	class EtherscanRateLimitError(EtherscanClientException):	4✔
21	pass	4✔
22
23
24	class EtherscanClient:	4✔
25	NETWORK_WITH_URL = {	4✔
26	EthereumNetwork.MAINNET: "https://etherscan.io",
27	EthereumNetwork.RINKEBY: "https://rinkeby.etherscan.io",
28	EthereumNetwork.ROPSTEN: "https://ropsten.etherscan.io",
29	EthereumNetwork.GOERLI: "https://goerli.etherscan.io",
30	EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://bscscan.com",
31	EthereumNetwork.POLYGON: "https://polygonscan.com",
32	EthereumNetwork.POLYGON_ZKEVM: "https://zkevm.polygonscan.com",
33	EthereumNetwork.OPTIMISM: "https://optimistic.etherscan.io",
34	EthereumNetwork.ARBITRUM_ONE: "https://arbiscan.io",
35	EthereumNetwork.ARBITRUM_NOVA: "https://nova.arbiscan.io",
36	EthereumNetwork.ARBITRUM_GOERLI: "https://goerli.arbiscan.io",
37	EthereumNetwork.AVALANCHE_C_CHAIN: "https://snowtrace.io",
38	EthereumNetwork.GNOSIS: "https://gnosisscan.io",
39	EthereumNetwork.MOONBEAM: "https://moonbeam.moonscan.io",
40	EthereumNetwork.MOONRIVER: "https://moonriver.moonscan.io",
41	EthereumNetwork.MOONBASE_ALPHA: "https://moonbase.moonscan.io",
42	EthereumNetwork.CRONOS_MAINNET: "https://cronoscan.com",
43	EthereumNetwork.CRONOS_TESTNET: "https://testnet.cronoscan.com",
44	EthereumNetwork.CELO_MAINNET: "https://celoscan.io",
45	EthereumNetwork.BASE_GOERLI_TESTNET: "https://goerli.basescan.org",
46	EthereumNetwork.NEON_EVM_DEVNET: "https://devnet.neonscan.org",
47	EthereumNetwork.NEON_EVM_MAINNET: "https://neonscan.org",
48	EthereumNetwork.SEPOLIA: "https://sepolia.etherscan.io",
49	EthereumNetwork.ZKSYNC_MAINNET: "https://explorer.zksync.io/",
50	EthereumNetwork.FANTOM_OPERA: "https://ftmscan.com",
51	EthereumNetwork.FANTOM_TESTNET: "https://testnet.ftmscan.com/",
52	EthereumNetwork.LINEA: "https://lineascan.build",
53	EthereumNetwork.LINEA_GOERLI: "https://goerli.lineascan.build",
54	EthereumNetwork.MANTLE: "https://explorer.mantle.xyz",
55	EthereumNetwork.MANTLE_TESTNET: "https://explorer.testnet.mantle.xyz",
56	EthereumNetwork.JAPAN_OPEN_CHAIN_MAINNET: "https://mainnet.japanopenchain.org",
57	EthereumNetwork.JAPAN_OPEN_CHAIN_TESTNET: "https://explorer.testnet.japanopenchain.org",
58	EthereumNetwork.SCROLL_SEPOLIA_TESTNET: "https://sepolia.scrollscan.dev",
59	EthereumNetwork.SCROLL: "https://scrollscan.com",
60	EthereumNetwork.KROMA: "https://kromascan.com",
61	EthereumNetwork.KROMA_SEPOLIA: "https://sepolia.kromascan.com",
62	EthereumNetwork.BLAST_SEPOLIA_TESTNET: "https://sepolia.blastscan.io",
63	EthereumNetwork.FRAXTAL: "https://fraxscan.com",
64	EthereumNetwork.BASE: "https://api.basescan.org/",
65	EthereumNetwork.BLAST: "https://blastscan.io",
66	EthereumNetwork.TAIKO_MAINNET: "https://taikoscan.io",
67	EthereumNetwork.BASE_SEPOLIA_TESTNET: "https://sepolia.basescan.org",
68	EthereumNetwork.HOLESKY: "https://holesky.etherscan.io",
69	EthereumNetwork.LINEA_SEPOLIA: "https://sepolia.lineascan.build",
70	}
71
72	NETWORK_WITH_API_URL = {	4✔
73	EthereumNetwork.MAINNET: "https://api.etherscan.io",
74	EthereumNetwork.RINKEBY: "https://api-rinkeby.etherscan.io",
75	EthereumNetwork.ROPSTEN: "https://api-ropsten.etherscan.io",
76	EthereumNetwork.GOERLI: "https://api-goerli.etherscan.io",
77	EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://api.bscscan.com",
78	EthereumNetwork.POLYGON: "https://api.polygonscan.com",
79	EthereumNetwork.POLYGON_ZKEVM: "https://api-zkevm.polygonscan.com",
80	EthereumNetwork.OPTIMISM: "https://api-optimistic.etherscan.io",
81	EthereumNetwork.ARBITRUM_ONE: "https://api.arbiscan.io",
82	EthereumNetwork.ARBITRUM_NOVA: "https://api-nova.arbiscan.io",
83	EthereumNetwork.ARBITRUM_GOERLI: "https://api-goerli.arbiscan.io",
84	EthereumNetwork.ARBITRUM_SEPOLIA: "https://api-sepolia.arbiscan.io",
85	EthereumNetwork.AVALANCHE_C_CHAIN: "https://api.snowtrace.io",
86	EthereumNetwork.GNOSIS: "https://api.gnosisscan.io",
87	EthereumNetwork.MOONBEAM: "https://api-moonbeam.moonscan.io",
88	EthereumNetwork.MOONRIVER: "https://api-moonriver.moonscan.io",
89	EthereumNetwork.MOONBASE_ALPHA: "https://api-moonbase.moonscan.io",
90	EthereumNetwork.CRONOS_MAINNET: "https://api.cronoscan.com",
91	EthereumNetwork.CRONOS_TESTNET: "https://api-testnet.cronoscan.com",
92	EthereumNetwork.CELO_MAINNET: "https://api.celoscan.io",
93	EthereumNetwork.BASE_GOERLI_TESTNET: "https://api-goerli.basescan.org",
94	EthereumNetwork.NEON_EVM_DEVNET: "https://devnet-api.neonscan.org",
95	EthereumNetwork.NEON_EVM_MAINNET: "https://api.neonscan.org",
96	EthereumNetwork.SEPOLIA: "https://api-sepolia.etherscan.io",
97	EthereumNetwork.ZKSYNC_MAINNET: "https://block-explorer-api.mainnet.zksync.io/",
98	EthereumNetwork.FANTOM_OPERA: "https://api.ftmscan.com",
99	EthereumNetwork.FANTOM_TESTNET: "https://api-testnet.ftmscan.com",
100	EthereumNetwork.LINEA: "https://api.lineascan.build",
101	EthereumNetwork.LINEA_GOERLI: "https://api-testnet.lineascan.build",
102	EthereumNetwork.MANTLE: "https://explorer.mantle.xyz",
103	EthereumNetwork.MANTLE_TESTNET: "https://explorer.testnet.mantle.xyz",
104	EthereumNetwork.JAPAN_OPEN_CHAIN_MAINNET: "https://mainnet.japanopenchain.org/api",
105	EthereumNetwork.JAPAN_OPEN_CHAIN_TESTNET: "https://explorer.testnet.japanopenchain.org/api",
106	EthereumNetwork.SCROLL_SEPOLIA_TESTNET: "https://api-sepolia.scrollscan.dev",
107	EthereumNetwork.SCROLL: "https://api.scrollscan.com",
108	EthereumNetwork.KROMA: "https://api.kromascan.com",
109	EthereumNetwork.KROMA_SEPOLIA: "https://api-sepolia.kromascan.com",
110	EthereumNetwork.BLAST_SEPOLIA_TESTNET: "https://api-sepolia.blastscan.io",
111	EthereumNetwork.FRAXTAL: "https://api.fraxscan.com",
112	EthereumNetwork.BASE: "https://api.basescan.org",
113	EthereumNetwork.BLAST: "https://api.blastscan.io",
114	EthereumNetwork.TAIKO_MAINNET: "https://api.taikoscan.io",
115	EthereumNetwork.BASE_SEPOLIA_TESTNET: "https://api-sepolia.basescan.org/api",
116	EthereumNetwork.HOLESKY: "https://api-holesky.etherscan.io",
117	EthereumNetwork.LINEA_SEPOLIA: "https://api-sepolia.lineascan.build",
118	}
119	HTTP_HEADERS: MutableMapping[str, Union[str, bytes]] = {	4✔
120	"User-Agent": "curl/7.77.0",
121	}
122
123	def __init__(	4✔
124	self,
125	network: EthereumNetwork,
126	api_key: Optional[str] = None,
127	request_timeout: int = int(
128	os.environ.get("ETHERSCAN_CLIENT_REQUEST_TIMEOUT", 10)
129	),
130	):
131	self.network = network	4✔
132	self.api_key = api_key	4✔
133	self.base_url = self.NETWORK_WITH_URL.get(network, "")	4✔
134	self.base_api_url = self.NETWORK_WITH_API_URL.get(network, "")	4✔
135	if not self.base_api_url:	4✔
136	raise EtherscanClientConfigurationProblem(	×
137	f"Network {network.name} - {network.value} not supported"
138	)
139	self.http_session = prepare_http_session(10, 100)	4✔
140	self.http_session.headers = self.HTTP_HEADERS	4✔
141	self.request_timeout = request_timeout	4✔
142
143	def build_url(self, path: str):	4✔
144	url = urljoin(self.base_api_url, path)	4✔
145	if self.api_key:	4✔
146	url += f"&apikey={self.api_key}"	4✔
147	return url	4✔
148
149	def _do_request(self, url: str) -> Optional[Union[Dict[str, Any], List[Any], str]]:	4✔
150	response = self.http_session.get(url, timeout=self.request_timeout)	4✔
151
152	if response.ok:	4✔
153	response_json = response.json()	4✔
154	result = response_json["result"]	4✔
155	if "Max rate limit reached" in result:	4✔
156	# Max rate limit reached, please use API Key for higher rate limit
157	raise EtherscanRateLimitError	×
158	if response_json["status"] == "1":	4✔
159	return result	4✔
160	return None	4✔
161
162	def _retry_request(	4✔
163	self, url: str, retry: bool = True
164	) -> Optional[Union[Dict[str, Any], List[Any], str]]:
165	for _ in range(3):	4✔
166	try:	4✔
167	return self._do_request(url)	4✔
168	except EtherscanRateLimitError as exc:	×
169	if not retry:	×
170	raise exc	×
171	else:
172	time.sleep(5)	×
173	return None	×
174
175	def get_contract_metadata(	4✔
176	self, contract_address: str, retry: bool = True
177	) -> Optional[ContractMetadata]:
178	contract_source_code = self.get_contract_source_code(	4✔
179	contract_address, retry=retry
180	)
181	if contract_source_code:	4✔
182	contract_name = contract_source_code["ContractName"]	4✔
183	contract_abi = contract_source_code["ABI"]	4✔
184	if contract_abi:	4✔
185	return ContractMetadata(contract_name, contract_abi, False)	4✔
186	return None	4✔
187
188	def get_contract_source_code(self, contract_address: str, retry: bool = True):	4✔
189	"""
190	Get source code for a contract. Source code query also returns:
191
192	- ContractName: "",
193	- CompilerVersion: "",
194	- OptimizationUsed: "",
195	- Runs: "",
196	- ConstructorArguments: ""
197	- EVMVersion: "Default",
198	- Library: "",
199	- LicenseType: "",
200	- Proxy: "0",
201	- Implementation: "",
202	- SwarmSource: ""
203
204	:param contract_address:
205	:param retry: if ``True``, try again if there's Rate Limit Error
206	:return:
207	"""
208	url = self.build_url(	4✔
209	f"api?module=contract&action=getsourcecode&address={contract_address}"
210	)
211	response = self._retry_request(url, retry=retry) # Returns a list	4✔
212	if response and isinstance(response, list):	4✔
213	result = response[0]	4✔
214	abi_str = result.get("ABI")	4✔
215
216	if isinstance(abi_str, str) and abi_str.startswith("["):	4✔
217	try:	4✔
218	result["ABI"] = json.loads(abi_str)	4✔
219	except json.JSONDecodeError:	×
220	result["ABI"] = None # Handle the case where JSON decoding fails	×
221	else:
222	result["ABI"] = None	4✔
223
224	return result	4✔
225
226	def get_contract_abi(self, contract_address: str, retry: bool = True):	4✔
227	url = self.build_url(	4✔
228	f"api?module=contract&action=getabi&address={contract_address}"
229	)
230	result = self._retry_request(url, retry=retry)	4✔
231	if isinstance(result, dict):	4✔
232	return result	×
233	elif isinstance(result, str):	4✔
234	try:	4✔
235	return json.loads(result)	4✔
236	except json.JSONDecodeError:	×
237	pass	×
238	return None	4✔

safe-global / safe-eth-py / 10793540350

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous