Move abis to abis folder

- Rename `deploy_safe_master_copy` and `deploy_proxy_factory_contract` to `deploy_contract`

gnosis/eth/contracts/__init__.py

@@ -1,3 +1,4 @@
+# flake8: noqa F401
 """
 Safe Addresses. Should be the same for every chain except for the ones with `chainId` protection. Check:
 https://github.com/safe-global/safe-deployments/tree/main/src/assets
@@ -32,13 +33,16 @@
 
 from gnosis.util import cache
 
+from .abis.multicall import multicall_v3_abi, multicall_v3_bytecode
+from .contract_base import ContractBase
+
 
 def load_contract_interface(file_name):
     return _load_json_file(_abi_file_path(file_name))
 
 
 def _abi_file_path(file):
-    return os.path.abspath(os.path.join(os.path.dirname(__file__), file))
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), "abis", file))
 
 
 def _load_json_file(path):
@@ -91,7 +95,7 @@ def fn(w3: Web3, address: Optional[ChecksumAddress] = None):
 
 
 # Anotate functions that will be generated later with `setattr` so typing does not complains
-def get_safe_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_safe_contract(w3: Web3, address: Optional[ChecksumAddress] = None) -> Contract:
     """
     :param w3:
     :param address:
@@ -100,116 +104,154 @@ def get_safe_contract(w3: Web3, address: Optional[str] = None) -> Contract:
     return get_safe_V1_3_0_contract(w3, address=address)
 
 
-def get_safe_V1_3_0_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_safe_V1_3_0_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_safe_V1_1_1_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_safe_V1_1_1_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_safe_V1_0_0_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_safe_V1_0_0_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_safe_V0_0_1_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_safe_V0_0_1_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
 def get_compatibility_fallback_handler_V1_3_0_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
-def get_erc20_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_erc20_contract(w3: Web3, address: Optional[ChecksumAddress] = None) -> Contract:
     pass
 
 
-def get_erc721_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_erc721_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_erc1155_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_erc1155_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_example_erc20_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_example_erc20_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
 def get_delegate_constructor_proxy_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
-def get_multi_send_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_multi_send_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_paying_proxy_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_paying_proxy_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_proxy_factory_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_proxy_factory_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     return get_proxy_factory_V1_3_0_contract(w3, address)
 
 
 def get_proxy_factory_V1_3_0_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
 def get_proxy_factory_V1_1_1_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
 def get_proxy_factory_V1_0_0_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
-def get_proxy_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_proxy_contract(w3: Web3, address: Optional[ChecksumAddress] = None) -> Contract:
     pass
 
 
-def get_uniswap_exchange_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_uniswap_exchange_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_uniswap_factory_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_uniswap_factory_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
 def get_uniswap_v2_factory_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
-def get_uniswap_v2_pair_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_uniswap_v2_pair_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
-def get_uniswap_v2_router_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_uniswap_v2_router_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
 def get_kyber_network_proxy_contract(
-    w3: Web3, address: Optional[str] = None
+    w3: Web3, address: Optional[ChecksumAddress] = None
 ) -> Contract:
     pass
 
 
-def get_cpk_factory_contract(w3: Web3, address: Optional[str] = None) -> Contract:
+def get_cpk_factory_contract(
+    w3: Web3, address: Optional[ChecksumAddress] = None
+) -> Contract:
     pass
 
 
+def get_multicall_v3_contract(w3: Web3, address: Optional[ChecksumAddress] = None):
+    return w3.eth.contract(
+        address=address,
+        abi=multicall_v3_abi,
+        bytecode=multicall_v3_bytecode,
+    )
+
+
 @cache
 def get_proxy_1_3_0_deployed_bytecode() -> bytes:
     return HexBytes(load_contract_interface("Proxy_V1_3_0.json")["deployedBytecode"])

gnosis/eth/contracts/contract_base.py

@@ -0,0 +1,30 @@
+from abc import ABCMeta, abstractmethod
+from functools import cached_property
+from logging import getLogger
+from typing import Callable, Optional
+
+from eth_typing import ChecksumAddress
+from web3 import Web3
+from web3.contract import Contract
+
+from .. import EthereumClient
+
+logger = getLogger(__name__)
+
+
+class ContractBase(metaclass=ABCMeta):
+    def __init__(self, address: ChecksumAddress, ethereum_client: EthereumClient):
+        self.address = address
+        self.ethereum_client = ethereum_client
+        self.w3 = ethereum_client.w3
+
+    @abstractmethod
+    def get_contract_fn(self) -> Callable[[Web3, Optional[ChecksumAddress]], Contract]:
+        """
+        :return: Contract function to get the proper contract
+        """
+        raise NotImplementedError
+
+    @cached_property
+    def contract(self) -> Contract:
+        return self.get_contract_fn()(self.ethereum_client.w3, self.address)

gnosis/eth/ethereum_client.py

@@ -1411,7 +1411,7 @@ def deploy_and_initialize_contract(
         constructor_data: bytes,
         initializer_data: bytes = b"",
         check_receipt: bool = True,
-    ):
+    ) -> EthereumTxSent:
         contract_address: Optional[ChecksumAddress] = None
         for data in (constructor_data, initializer_data):
             # Because initializer_data is not mandatory
@@ -1505,6 +1505,12 @@ def estimate_gas(
 
     @staticmethod
     def estimate_data_gas(data: bytes):
+        """
+        Estimate gas costs only for "storage" of the ``data`` bytes provided
+
+        :param data:
+        :return:
+        """
         if isinstance(data, str):
             data = HexBytes(data)
 
@@ -1698,6 +1704,57 @@ def get_blocks(
     def is_contract(self, contract_address: ChecksumAddress) -> bool:
         return bool(self.w3.eth.get_code(contract_address))
 
+    @staticmethod
+    def build_tx_params(
+        from_address: Optional[ChecksumAddress] = None,
+        to_address: Optional[ChecksumAddress] = None,
+        value: Optional[int] = None,
+        gas: Optional[int] = None,
+        gas_price: Optional[int] = None,
+        nonce: Optional[int] = None,
+        chain_id: Optional[int] = None,
+        tx_params: Optional[TxParams] = None,
+    ) -> TxParams:
+        """
+        Build tx params dictionary.
+        If an existing TxParams dictionary is provided the fields will be replaced by the provided ones
+
+        :param from_address:
+        :param to_address:
+        :param value:
+        :param gas:
+        :param gas_price:
+        :param nonce:
+        :param chain_id:
+        :param tx_params: An existing TxParams dictionary will be replaced by the providen values
+        :return:
+        """
+
+        tx_params: TxParams = tx_params or {}
+
+        if from_address:
+            tx_params["from"] = from_address
+
+        if to_address:
+            tx_params["to"] = to_address
+
+        if value is not None:
+            tx_params["value"] = value
+
+        if gas_price is not None:
+            tx_params["gasPrice"] = gas_price
+
+        if gas is not None:
+            tx_params["gas"] = gas
+
+        if nonce is not None:
+            tx_params["nonce"] = nonce
+
+        if chain_id is not None:
+            tx_params["chainId"] = chain_id
+
+        return tx_params
+
     @tx_with_exception_handling
     def send_transaction(self, transaction_dict: TxParams) -> HexBytes:
         return self.w3.eth.send_transaction(transaction_dict)