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

93.81

/safe_eth/eth/multicall.py

"""
MultiCall Smart Contract API
https://github.com/mds1/multicall
"""
import logging
from dataclasses import dataclass
from typing import Any, Callable, List, Optional, Sequence, Tuple

import eth_abi
from eth_abi.exceptions import DecodingError
from eth_account.signers.local import LocalAccount
from eth_typing import BlockIdentifier, BlockNumber, ChecksumAddress, HexAddress, HexStr
from hexbytes import HexBytes
from web3 import Web3
from web3._utils.abi import map_abi_data
from web3._utils.normalizers import BASE_RETURN_NORMALIZERS
from web3.contract.contract import Contract, ContractFunction
from web3.exceptions import ContractLogicError

from . import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported
from .contracts import ContractBase, get_multicall_v3_contract
from .ethereum_client import EthereumTxSent
from .exceptions import BatchCallFunctionFailed, ContractAlreadyDeployed
from .utils import get_empty_tx_params

logger = logging.getLogger(__name__)


@dataclass
class MulticallResult:
    success: bool
    return_data: Optional[bytes]


@dataclass
class MulticallDecodedResult:
    success: bool
    return_data_decoded: Optional[Any]


class Multicall(ContractBase):
    # https://github.com/mds1/multicall#deployments
    ADDRESSES = {
        EthereumNetwork.MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.GOERLI: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.SEPOLIA: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.OPTIMISM: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.OPTIMISM_GOERLI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.ARBITRUM_ONE: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.ARBITRUM_NOVA: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.ARBITRUM_GOERLI: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.POLYGON: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.MUMBAI: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.POLYGON_ZKEVM: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.POLYGON_ZKEVM_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.GNOSIS: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.AVALANCHE_C_CHAIN: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.AVALANCHE_FUJI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.FANTOM_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.FANTOM_OPERA: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.BNB_SMART_CHAIN_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.RINKEBY: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.KCC_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.KCC_TESTNET: "0x665683D9bd41C09cF38c3956c926D9924F1ADa97",
        EthereumNetwork.ROPSTEN: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.CELO_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.CELO_ALFAJORES_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.AURORA_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
        EthereumNetwork.BASE_GOERLI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
    }

    def __init__(
        self,
        ethereum_client: EthereumClient,
        multicall_contract_address: Optional[ChecksumAddress] = None,
    ):
        ethereum_network = ethereum_client.get_network()
        address = multicall_contract_address or self.ADDRESSES.get(ethereum_network)
        mainnet_address = self.ADDRESSES.get(EthereumNetwork.MAINNET)
        if not address and mainnet_address:
            # Try with Multicall V3 deterministic address
            address = ChecksumAddress(HexAddress(HexStr(mainnet_address)))
            if not ethereum_client.is_contract(address):
                raise EthereumNetworkNotSupported(
                    "Multicall contract not available for %s", ethereum_network.name
                )

        if not address:
            raise ValueError("Contract address cannot be none")

        super().__init__(ChecksumAddress(HexAddress(HexStr(address))), ethereum_client)

    def get_contract_fn(self) -> Callable[[Web3, Optional[ChecksumAddress]], Contract]:
        return get_multicall_v3_contract

    @classmethod
    def deploy_contract(
        cls, ethereum_client: EthereumClient, deployer_account: LocalAccount
    ) -> Optional[EthereumTxSent]:
        """
        Deploy contract

        :param ethereum_client:
        :param deployer_account: Ethereum Account
        :return: ``EthereumTxSent`` with the deployed contract address, ``None`` if already deployed
        """
        contract_fn = cls.get_contract_fn(cls)  # type: ignore[arg-type]
        contract = contract_fn(ethereum_client.w3, None)
        constructor_data = contract.constructor().build_transaction(
            get_empty_tx_params()
        )["data"]

        try:
            ethereum_tx_sent = ethereum_client.deploy_and_initialize_contract(
                deployer_account, constructor_data
            )

            assert ethereum_tx_sent.contract_address is not None
            contract_address = ethereum_tx_sent.contract_address
            logger.info(
                "Deployed Multicall V2 Contract %s by %s",
                contract_address,
                deployer_account.address,
            )
            # Add address to addresses dictionary
            cls.ADDRESSES[ethereum_client.get_network()] = contract_address
            return ethereum_tx_sent
        except ContractAlreadyDeployed as e:
            cls.ADDRESSES[ethereum_client.get_network()] = e.address
            return None

    @staticmethod
    def _build_payload(
        contract_functions: Sequence[ContractFunction],
    ) -> Tuple[List[Tuple[ChecksumAddress, HexBytes]], List[List[Any]]]:
        targets_with_data = []
        output_types = []
        for contract_function in contract_functions:
            targets_with_data.append(
                (
                    contract_function.address,
                    HexBytes(contract_function._encode_transaction_data()),
                )
            )
            output_types.append(
                [output["type"] for output in contract_function.abi["outputs"]]
            )

        return targets_with_data, output_types

    def _build_payload_same_function(
        self,
        contract_function: ContractFunction,
        contract_addresses: Sequence[ChecksumAddress],
    ) -> Tuple[List[Tuple[ChecksumAddress, HexBytes]], List[List[Any]]]:
        targets_with_data = []
        output_types = []
        tx_data = HexBytes(contract_function._encode_transaction_data())
        for contract_address in contract_addresses:
            targets_with_data.append((contract_address, tx_data))
            output_types.append(
                [output["type"] for output in contract_function.abi["outputs"]]
            )

        return targets_with_data, output_types

    def _decode_data(self, output_type: Sequence[str], data: bytes) -> Optional[Any]:
        """

        :param output_type:
        :param data:
        :return:
        :raises: DecodingError
        """
        if data:
            try:
                decoded_values = eth_abi.decode(output_type, data)
                normalized_data = map_abi_data(
                    BASE_RETURN_NORMALIZERS, output_type, decoded_values
                )
                if len(normalized_data) == 1:
                    return normalized_data[0]
                else:
                    return normalized_data
            except DecodingError:
                logger.warning(
                    "Cannot decode %s using output-type %s", data, output_type
                )
                return data
        return None

    def _aggregate(
        self,
        targets_with_data: Sequence[Tuple[ChecksumAddress, bytes]],
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> Tuple[BlockNumber, List[Optional[Any]]]:
        """

        :param targets_with_data: List of target `addresses` and `data` to be called in each Contract
        :param block_identifier:
        :return:
        :raises: BatchCallFunctionFailed
        """
        aggregate_parameter = [
            {"target": target, "callData": data} for target, data in targets_with_data
        ]
        try:
            return self.contract.functions.aggregate(aggregate_parameter).call(
                block_identifier=block_identifier or "latest"
            )
        except (ContractLogicError, OverflowError):
            raise BatchCallFunctionFailed

    def aggregate(
        self,
        contract_functions: Sequence[ContractFunction],
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> Tuple[BlockNumber, List[Optional[Any]]]:
        """
        Calls ``aggregate`` on MakerDAO's Multicall contract. If a function called raises an error execution is stopped

        :param contract_functions:
        :param block_identifier:
        :return: A tuple with the ``blockNumber`` and a list with the decoded return values
        :raises: BatchCallFunctionFailed
        """
        targets_with_data, output_types = self._build_payload(contract_functions)
        block_number, results = self._aggregate(
            targets_with_data, block_identifier=block_identifier
        )
        decoded_results = [
            self._decode_data(output_type, data) if data is not None else None
            for output_type, data in zip(output_types, results)
        ]
        return block_number, decoded_results

    def _try_aggregate(
        self,
        targets_with_data: Sequence[Tuple[ChecksumAddress, bytes]],
        require_success: bool = False,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> List[MulticallResult]:
        """
        Calls ``try_aggregate`` on MakerDAO's Multicall contract.

        :param targets_with_data:
        :param require_success: If ``True``, an exception in any of the functions will stop the execution. Also, an
            invalid decoded value will stop the execution
        :param block_identifier:
        :return: A list with the decoded return values
        """

        aggregate_parameter = [
            {"target": target, "callData": data} for target, data in targets_with_data
        ]
        try:
            result = self.contract.functions.tryAggregate(
                require_success, aggregate_parameter
            ).call(block_identifier=block_identifier or "latest")

            if require_success and b"" in (data for _, data in result):
                # `b''` values are decoding errors/missing contracts/missing functions
                raise BatchCallFunctionFailed

            return [
                MulticallResult(success, data if data else None)
                for success, data in result
            ]
        except (ContractLogicError, OverflowError, ValueError):
            raise BatchCallFunctionFailed

    def try_aggregate(
        self,
        contract_functions: Sequence[ContractFunction],
        require_success: bool = False,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> List[MulticallDecodedResult]:
        """
        Calls ``try_aggregate`` on MakerDAO's Multicall contract.

        :param contract_functions:
        :param require_success: If ``True``, an exception in any of the functions will stop the execution
        :param block_identifier:
        :return: A list with the decoded return values
        """
        targets_with_data, output_types = self._build_payload(contract_functions)
        results = self._try_aggregate(
            targets_with_data,
            require_success=require_success,
            block_identifier=block_identifier,
        )
        return [
            MulticallDecodedResult(
                multicall_result.success,
                self._decode_data(output_type, multicall_result.return_data)
                if multicall_result.success and multicall_result.return_data is not None
                else multicall_result.return_data,
            )
            for output_type, multicall_result in zip(output_types, results)
        ]

    def try_aggregate_same_function(
        self,
        contract_function: ContractFunction,
        contract_addresses: Sequence[ChecksumAddress],
        require_success: bool = False,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> List[MulticallDecodedResult]:
        """
        Calls ``try_aggregate`` on MakerDAO's Multicall contract. Reuse same function with multiple contract addresses.
        It's more optimal due to instantiating ``ContractFunction`` objects is very demanding

        :param contract_function:
        :param contract_addresses:
        :param require_success: If ``True``, an exception in any of the functions will stop the execution
        :param block_identifier:
        :return: A list with the decoded return values
        """

        targets_with_data, output_types = self._build_payload_same_function(
            contract_function, contract_addresses
        )
        results = self._try_aggregate(
            targets_with_data,
            require_success=require_success,
            block_identifier=block_identifier,
        )
        return [
            MulticallDecodedResult(
                multicall_result.success,
                self._decode_data(output_type, multicall_result.return_data)
                if multicall_result.success and multicall_result.return_data is not None
                else multicall_result.return_data,
            )
            for output_type, multicall_result in zip(output_types, results)
        ]

1	"""
2	MultiCall Smart Contract API
3	https://github.com/mds1/multicall
4	"""
5	import logging	4✔
6	from dataclasses import dataclass	4✔
7	from typing import Any, Callable, List, Optional, Sequence, Tuple	4✔
8
9	import eth_abi	4✔
10	from eth_abi.exceptions import DecodingError	4✔
11	from eth_account.signers.local import LocalAccount	4✔
12	from eth_typing import BlockIdentifier, BlockNumber, ChecksumAddress, HexAddress, HexStr	4✔
13	from hexbytes import HexBytes	4✔
14	from web3 import Web3	4✔
15	from web3._utils.abi import map_abi_data	4✔
16	from web3._utils.normalizers import BASE_RETURN_NORMALIZERS	4✔
17	from web3.contract.contract import Contract, ContractFunction	4✔
18	from web3.exceptions import ContractLogicError	4✔
19
20	from . import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported	4✔
21	from .contracts import ContractBase, get_multicall_v3_contract	4✔
22	from .ethereum_client import EthereumTxSent	4✔
23	from .exceptions import BatchCallFunctionFailed, ContractAlreadyDeployed	4✔
24	from .utils import get_empty_tx_params	4✔
25
26	logger = logging.getLogger(__name__)	4✔
27
28
29	@dataclass	4✔
30	class MulticallResult:	4✔
31	success: bool	4✔
32	return_data: Optional[bytes]	4✔
33
34
35	@dataclass	4✔
36	class MulticallDecodedResult:	4✔
37	success: bool	4✔
38	return_data_decoded: Optional[Any]	4✔
39
40
41	class Multicall(ContractBase):	4✔
42	# https://github.com/mds1/multicall#deployments
43	ADDRESSES = {	4✔
44	EthereumNetwork.MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
45	EthereumNetwork.GOERLI: "0xcA11bde05977b3631167028862bE2a173976CA11",
46	EthereumNetwork.SEPOLIA: "0xcA11bde05977b3631167028862bE2a173976CA11",
47	EthereumNetwork.OPTIMISM: "0xcA11bde05977b3631167028862bE2a173976CA11",
48	EthereumNetwork.OPTIMISM_GOERLI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
49	EthereumNetwork.ARBITRUM_ONE: "0xcA11bde05977b3631167028862bE2a173976CA11",
50	EthereumNetwork.ARBITRUM_NOVA: "0xcA11bde05977b3631167028862bE2a173976CA11",
51	EthereumNetwork.ARBITRUM_GOERLI: "0xcA11bde05977b3631167028862bE2a173976CA11",
52	EthereumNetwork.POLYGON: "0xcA11bde05977b3631167028862bE2a173976CA11",
53	EthereumNetwork.MUMBAI: "0xcA11bde05977b3631167028862bE2a173976CA11",
54	EthereumNetwork.POLYGON_ZKEVM: "0xcA11bde05977b3631167028862bE2a173976CA11",
55	EthereumNetwork.POLYGON_ZKEVM_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
56	EthereumNetwork.GNOSIS: "0xcA11bde05977b3631167028862bE2a173976CA11",
57	EthereumNetwork.AVALANCHE_C_CHAIN: "0xcA11bde05977b3631167028862bE2a173976CA11",
58	EthereumNetwork.AVALANCHE_FUJI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
59	EthereumNetwork.FANTOM_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
60	EthereumNetwork.FANTOM_OPERA: "0xcA11bde05977b3631167028862bE2a173976CA11",
61	EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
62	EthereumNetwork.BNB_SMART_CHAIN_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
63	EthereumNetwork.RINKEBY: "0xcA11bde05977b3631167028862bE2a173976CA11",
64	EthereumNetwork.KCC_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
65	EthereumNetwork.KCC_TESTNET: "0x665683D9bd41C09cF38c3956c926D9924F1ADa97",
66	EthereumNetwork.ROPSTEN: "0xcA11bde05977b3631167028862bE2a173976CA11",
67	EthereumNetwork.CELO_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
68	EthereumNetwork.CELO_ALFAJORES_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
69	EthereumNetwork.AURORA_MAINNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
70	EthereumNetwork.BASE_GOERLI_TESTNET: "0xcA11bde05977b3631167028862bE2a173976CA11",
71	}
72
73	def __init__(	4✔
74	self,
75	ethereum_client: EthereumClient,
76	multicall_contract_address: Optional[ChecksumAddress] = None,
77	):
78	ethereum_network = ethereum_client.get_network()	4✔
79	address = multicall_contract_address or self.ADDRESSES.get(ethereum_network)	4✔
80	mainnet_address = self.ADDRESSES.get(EthereumNetwork.MAINNET)	4✔
81	if not address and mainnet_address:	4✔
82	# Try with Multicall V3 deterministic address
83	address = ChecksumAddress(HexAddress(HexStr(mainnet_address)))	4✔
84	if not ethereum_client.is_contract(address):	4✔
85	raise EthereumNetworkNotSupported(	4✔
86	"Multicall contract not available for %s", ethereum_network.name
87	)
88
89	if not address:	4✔
90	raise ValueError("Contract address cannot be none")	×
91
92	super().__init__(ChecksumAddress(HexAddress(HexStr(address))), ethereum_client)	4✔
93
94	def get_contract_fn(self) -> Callable[[Web3, Optional[ChecksumAddress]], Contract]:	4✔
95	return get_multicall_v3_contract	4✔
96
97	@classmethod	4✔
98	def deploy_contract(	4✔
99	cls, ethereum_client: EthereumClient, deployer_account: LocalAccount
100	) -> Optional[EthereumTxSent]:
101	"""
102	Deploy contract
103
104	:param ethereum_client:
105	:param deployer_account: Ethereum Account
106	:return: ``EthereumTxSent`` with the deployed contract address, ``None`` if already deployed
107	"""
108	contract_fn = cls.get_contract_fn(cls) # type: ignore[arg-type]	4✔
109	contract = contract_fn(ethereum_client.w3, None)	4✔
110	constructor_data = contract.constructor().build_transaction(	4✔
111	get_empty_tx_params()
112	)["data"]
113
114	try:	4✔
115	ethereum_tx_sent = ethereum_client.deploy_and_initialize_contract(	4✔
116	deployer_account, constructor_data
117	)
118
119	assert ethereum_tx_sent.contract_address is not None	4✔
120	contract_address = ethereum_tx_sent.contract_address	4✔
121	logger.info(	4✔
122	"Deployed Multicall V2 Contract %s by %s",
123	contract_address,
124	deployer_account.address,
125	)
126	# Add address to addresses dictionary
127	cls.ADDRESSES[ethereum_client.get_network()] = contract_address	4✔
128	return ethereum_tx_sent	4✔
129	except ContractAlreadyDeployed as e:	×
130	cls.ADDRESSES[ethereum_client.get_network()] = e.address	×
131	return None	×
132
133	@staticmethod	4✔
134	def _build_payload(	4✔
135	contract_functions: Sequence[ContractFunction],
136	) -> Tuple[List[Tuple[ChecksumAddress, HexBytes]], List[List[Any]]]:
137	targets_with_data = []	4✔
138	output_types = []	4✔
139	for contract_function in contract_functions:	4✔
140	targets_with_data.append(	4✔
141	(
142	contract_function.address,
143	HexBytes(contract_function._encode_transaction_data()),
144	)
145	)
146	output_types.append(	4✔
147	[output["type"] for output in contract_function.abi["outputs"]]
148	)
149
150	return targets_with_data, output_types	4✔
151
152	def _build_payload_same_function(	4✔
153	self,
154	contract_function: ContractFunction,
155	contract_addresses: Sequence[ChecksumAddress],
156	) -> Tuple[List[Tuple[ChecksumAddress, HexBytes]], List[List[Any]]]:
157	targets_with_data = []	4✔
158	output_types = []	4✔
159	tx_data = HexBytes(contract_function._encode_transaction_data())	4✔
160	for contract_address in contract_addresses:	4✔
161	targets_with_data.append((contract_address, tx_data))	4✔
162	output_types.append(	4✔
163	[output["type"] for output in contract_function.abi["outputs"]]
164	)
165
166	return targets_with_data, output_types	4✔
167
168	def _decode_data(self, output_type: Sequence[str], data: bytes) -> Optional[Any]:	4✔
169	"""
170
171	:param output_type:
172	:param data:
173	:return:
174	:raises: DecodingError
175	"""
176	if data:	4✔
177	try:	4✔
178	decoded_values = eth_abi.decode(output_type, data)	4✔
179	normalized_data = map_abi_data(	4✔
180	BASE_RETURN_NORMALIZERS, output_type, decoded_values
181	)
182	if len(normalized_data) == 1:	4✔
183	return normalized_data[0]	4✔
184	else:
185	return normalized_data	4✔
186	except DecodingError:	×
187	logger.warning(	×
188	"Cannot decode %s using output-type %s", data, output_type
189	)
190	return data	×
191	return None	4✔
192
193	def _aggregate(	4✔
194	self,
195	targets_with_data: Sequence[Tuple[ChecksumAddress, bytes]],
196	block_identifier: Optional[BlockIdentifier] = "latest",
197	) -> Tuple[BlockNumber, List[Optional[Any]]]:
198	"""
199
200	:param targets_with_data: List of target `addresses` and `data` to be called in each Contract
201	:param block_identifier:
202	:return:
203	:raises: BatchCallFunctionFailed
204	"""
205	aggregate_parameter = [	4✔
206	{"target": target, "callData": data} for target, data in targets_with_data
207	]
208	try:	4✔
209	return self.contract.functions.aggregate(aggregate_parameter).call(	4✔
210	block_identifier=block_identifier or "latest"
211	)
212	except (ContractLogicError, OverflowError):	4✔
213	raise BatchCallFunctionFailed	4✔
214
215	def aggregate(	4✔
216	self,
217	contract_functions: Sequence[ContractFunction],
218	block_identifier: Optional[BlockIdentifier] = "latest",
219	) -> Tuple[BlockNumber, List[Optional[Any]]]:
220	"""
221	Calls ``aggregate`` on MakerDAO's Multicall contract. If a function called raises an error execution is stopped
222
223	:param contract_functions:
224	:param block_identifier:
225	:return: A tuple with the ``blockNumber`` and a list with the decoded return values
226	:raises: BatchCallFunctionFailed
227	"""
228	targets_with_data, output_types = self._build_payload(contract_functions)	4✔
229	block_number, results = self._aggregate(	4✔
230	targets_with_data, block_identifier=block_identifier
231	)
232	decoded_results = [	4✔
233	self._decode_data(output_type, data) if data is not None else None
234	for output_type, data in zip(output_types, results)
235	]
236	return block_number, decoded_results	4✔
237
238	def _try_aggregate(	4✔
239	self,
240	targets_with_data: Sequence[Tuple[ChecksumAddress, bytes]],
241	require_success: bool = False,
242	block_identifier: Optional[BlockIdentifier] = "latest",
243	) -> List[MulticallResult]:
244	"""
245	Calls ``try_aggregate`` on MakerDAO's Multicall contract.
246
247	:param targets_with_data:
248	:param require_success: If ``True``, an exception in any of the functions will stop the execution. Also, an
249	invalid decoded value will stop the execution
250	:param block_identifier:
251	:return: A list with the decoded return values
252	"""
253
254	aggregate_parameter = [	4✔
255	{"target": target, "callData": data} for target, data in targets_with_data
256	]
257	try:	4✔
258	result = self.contract.functions.tryAggregate(	4✔
259	require_success, aggregate_parameter
260	).call(block_identifier=block_identifier or "latest")
261
262	if require_success and b"" in (data for _, data in result):	4✔
263	# `b''` values are decoding errors/missing contracts/missing functions
264	raise BatchCallFunctionFailed	4✔
265
266	return [	4✔
267	MulticallResult(success, data if data else None)
268	for success, data in result
269	]
270	except (ContractLogicError, OverflowError, ValueError):	4✔
271	raise BatchCallFunctionFailed	4✔
272
273	def try_aggregate(	4✔
274	self,
275	contract_functions: Sequence[ContractFunction],
276	require_success: bool = False,
277	block_identifier: Optional[BlockIdentifier] = "latest",
278	) -> List[MulticallDecodedResult]:
279	"""
280	Calls ``try_aggregate`` on MakerDAO's Multicall contract.
281
282	:param contract_functions:
283	:param require_success: If ``True``, an exception in any of the functions will stop the execution
284	:param block_identifier:
285	:return: A list with the decoded return values
286	"""
287	targets_with_data, output_types = self._build_payload(contract_functions)	4✔
288	results = self._try_aggregate(	4✔
289	targets_with_data,
290	require_success=require_success,
291	block_identifier=block_identifier,
292	)
293	return [	4✔
294	MulticallDecodedResult(
295	multicall_result.success,
296	self._decode_data(output_type, multicall_result.return_data)
297	if multicall_result.success and multicall_result.return_data is not None
298	else multicall_result.return_data,
299	)
300	for output_type, multicall_result in zip(output_types, results)
301	]
302
303	def try_aggregate_same_function(	4✔
304	self,
305	contract_function: ContractFunction,
306	contract_addresses: Sequence[ChecksumAddress],
307	require_success: bool = False,
308	block_identifier: Optional[BlockIdentifier] = "latest",
309	) -> List[MulticallDecodedResult]:
310	"""
311	Calls ``try_aggregate`` on MakerDAO's Multicall contract. Reuse same function with multiple contract addresses.
312	It's more optimal due to instantiating ``ContractFunction`` objects is very demanding
313
314	:param contract_function:
315	:param contract_addresses:
316	:param require_success: If ``True``, an exception in any of the functions will stop the execution
317	:param block_identifier:
318	:return: A list with the decoded return values
319	"""
320
321	targets_with_data, output_types = self._build_payload_same_function(	4✔
322	contract_function, contract_addresses
323	)
324	results = self._try_aggregate(	4✔
325	targets_with_data,
326	require_success=require_success,
327	block_identifier=block_identifier,
328	)
329	return [	4✔
330	MulticallDecodedResult(
331	multicall_result.success,
332	self._decode_data(output_type, multicall_result.return_data)
333	if multicall_result.success and multicall_result.return_data is not None
334	else multicall_result.return_data,
335	)
336	for output_type, multicall_result in zip(output_types, results)
337	]

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