update app scripts

Author: matthiaszimmermann

README.MD

@@ -366,8 +366,8 @@ forge script script/MockSetup.s.sol --fork-url http://127.0.0.1:8545 --broadcast
 
 ```shell
 cd app
-eval $(echo export $(grep MNEMONIC .env))
-eval $(echo export $(grep CONTRACT_ADDRESS .env))
+eval $(echo export $(grep PRODUCT_CONTRACT_ADDRESS .env))
+eval $(echo export $(grep OPERATOR_WALLET_MNEMONIC .env))
 uv run python
 ```
 
@@ -377,6 +377,7 @@ Follow the steps in the python shell below.
 import os
 
 from web3 import Web3
+from web3utils.chain import Chain
 from web3utils.contract import Contract
 from web3utils.wallet import Wallet
 
@@ -385,8 +386,8 @@ w3_uri = "http://127.0.0.1:8545"
 w3 = Web3(Web3.HTTPProvider(w3_uri))
 
 # create account using the env variable mnemonic
-w = Wallet.from_mnemonic(os.getenv('ETH_MNEMONIC'))
-operator = Wallet.from_mnemonic(os.getenv('OPERATOR_WALLET_MNEMONIC'))
+w = Wallet.from_mnemonic(os.getenv('ETH_MNEMONIC'), w3=w3)
+operator = Wallet.from_mnemonic(os.getenv('OPERATOR_WALLET_MNEMONIC'), w3=w3o)
 
 # send 1 eth to operator
 w.transfer(w3, operator.address, 1 * 10**18)

app/web3utils/chain.py

@@ -0,0 +1,38 @@
+
+from web3 import Web3
+
+class Chain:
+    """Simple class to query network."""
+
+    w3:Web3|None = None
+
+    def __init__(self, w3:Web3) -> None:
+        """Create a new chain query object.
+        """
+        self.w3 = w3
+
+    def w3(self) -> Web3:
+        """Get the web3 instance."""
+        if not self.w3:
+            raise ValueError("Web3 instance not provided")
+        return self.w3
+
+    def id(self) -> int:
+        """Get the network ID."""
+        chain_id = self.w3.net.version
+        try:
+            return int(chain_id)
+        except Exception as e:
+            raise ValueError(f"Chain Id {chain_id} is not an integer. Error: {e}")
+
+    def latest_block(self) -> int:
+        """Get the number of the latest block."""
+        self.w3.eth.block_number
+
+    def block(self, block_id='latest') -> dict:
+        """Get a block by its ID."""
+        return dict(self.w3.eth.get_block(block_id))
+
+    def timestamp(self, block_number='latest') -> int:
+        """Get the timestamp of a block."""
+        return self.w3.eth.get_block(block_number)['timestamp']

app/web3utils/contract.py

@@ -1,19 +1,19 @@
 import json
+import logging
+
+from functools import wraps
 from typing import Any, Dict
-from loguru import logger
 
 from web3 import Web3
 from web3.contract import Contract as Web3Contract
 from web3.exceptions import TimeExhausted
 from web3.types import FilterParams
-from util.logging import get_logger
-from web3utils.wallet import Wallet
 
-logger = get_logger()
+from web3utils.wallet import Wallet
 
 class Contract:
 
-    FOUNDRY_OUT = "./abi"
+    FOUNDRY_OUT = "../out"
     SOLIDITY_EXT = "sol"
     GAS = 1000000
     TX_TIMEOUT_SECONDS = 120
@@ -51,10 +51,10 @@ def _setup_functions(self) -> None:
                 mutability = func.get('stateMutability')
 
                 if mutability in ['nonpayable', 'payable']:
-                    logger.info(f"creating {func_name}(...) tx")
+                    logging.info(f"creating {func_name}(...) tx")
                     setattr(self, func_name, self._create_write_method(func_name))
                 elif mutability in ['view', 'pure']:
-                    logger.info(f"creating {func_name}(...) call")
+                    logging.info(f"creating {func_name}(...) call")
                     setattr(self, func_name, self._create_read_method(func_name))
 
     def _create_read_method(self, func_name: str):
@@ -63,14 +63,26 @@ def read_method(*args, **kwargs) -> Any:
                 modified_args = [arg.address if isinstance(arg, Wallet) else arg for arg in args]
                 return getattr(self.contract.functions, func_name)(*modified_args, **kwargs).call()
             except Exception as e:
-                logger.warning(f"Error calling function '{func_name}': {e}")
+                logging.warning(f"Error calling function '{func_name}': {e}")
                 return None
 
-        # Optionally, add docstrings or additional attributes here
-        read_method.__name__ = func_name
-        read_method.__doc__ = f"Calls the '{func_name}' function of the contract."
+        # add docstrings signature and selector
+        self._amend_method(read_method, func_name)
+
         return read_method
 
+    def _amend_method(self, method, name):
+        method.__name__ = name
+        method.__doc__ = f"Calls the '{name}' contract function."
+
+        signature = getattr(self.contract.functions, name).signature
+        method.signature = signature
+        method.argument_names = getattr(self.contract.functions, name).argument_names
+        method.inputs = getattr(self.contract.functions, name).abi['inputs']
+        method.outputs = getattr(self.contract.functions, name).abi['outputs']
+        method.selector = Web3.keccak(text=signature)[:4]
+        method.selector_hex = method.selector.hex()
+
     def _get_tx_params(self, args:tuple) -> Dict[str, Any]:
         if len(args) == 0:
             raise ValueError("No transaction parameters provided.")
@@ -95,38 +107,31 @@ def write_method(*args) -> str:
 
             try:
                 wallet = tx_params['from']
-                logger.info(f"Sending transaction for function '{func_name}' with wallet: {wallet.address}")
 
                 # create tx properties
                 chain_id = self.w3.eth.chain_id
                 gas = tx_params.get('gas', self.GAS)
                 gas_price = tx_params.get('gasPrice', self.w3.eth.gas_price)
-                gas_limit = tx_params.get('gasLimit', None)
                 nonce = self.w3.eth.get_transaction_count(wallet.address)
 
                 # transform wallet args to addresses (str)
                 modified_args = [arg.address if isinstance(arg, Wallet) else arg for arg in function_args]
 
-                options = {
+                # create tx
+                txn = getattr(self.contract.functions, func_name)(*modified_args).build_transaction({
                     'chainId': chain_id,
                     'gas': gas,
                     'gasPrice': gas_price,
                     'nonce': nonce,
-                }
-
-                if gas_limit:
-                    options['gas'] = gas_limit
-
-                # create tx
-                txn = getattr(self.contract.functions, func_name)(*modified_args).build_transaction(options)
+                })
 
                 # sign tx
                 private_key = bytes(wallet.account.key)
                 signed_txn = self.w3.eth.account.sign_transaction(txn, private_key=private_key)
 
                 # send signed transaction
                 tx_hash = self.w3.eth.send_raw_transaction(signed_txn.raw_transaction)
-                logger.info(f"Transaction sent: {tx_hash.hex()}")
+                logging.info(f"Transaction sent: {tx_hash.hex()}")
 
                 if 'timeout' not in tx_params:
                     timeout = self.TX_TIMEOUT_SECONDS
@@ -136,25 +141,23 @@ def write_method(*args) -> str:
                 try:
                     receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)
                 except TimeExhausted:
-                    logger.warning(f"Transaction timeout after {timeout} seconds.")
+                    logging.warning(f"Transaction timeout after {timeout} seconds.")
                     return tx_hash.hex()
 
                 if receipt['status'] == 1:
-                    logger.info(f"Transaction successful: {receipt}")
+                    logging.info(f"Transaction successful: {receipt}")
                 else:
-                    logger.warning(f"Transaction failed: {receipt}")
+                    logging.warning(f"Transaction failed: {receipt}")
 
                 return tx_hash.hex()
 
             except Exception as e:
-                logger.warning(f"Error sending transaction for function '{func_name}': {e}")
-                if 'tx_hash' in locals():
-                    logger.warning(f"Transaction hash: {tx_hash.hex()}")
-                    return tx_hash.hex()
-                return None
+                logging.warning(f"Error sending transaction for function '{func_name}': {e}")
+                return tx_hash.hex()
+
+        # add docstrings signature and selector
+        self._amend_method(write_method, func_name)
 
-        write_method.__name__ = func_name
-        write_method.__doc__ = f"Sends a transaction to the '{func_name}' function of the contract."
         return write_method
 
     def _load_abi(self, contract:str, out_path:str) -> Dict[str, Any]:

app/web3utils/wallet.py

@@ -1,13 +1,14 @@
-from typing import Any
+from typing import Any, Union
 
 from eth_account import Account
 from eth_account.hdaccount import ETHEREUM_DEFAULT_PATH
 from eth_account.signers.local import LocalAccount
 from eth_account.types import Language
 
-from util.password import generate_password
 from web3 import Web3
 
+from util.password import generate_password
+
 
 class Wallet:
     """Simple class for wallet creation."""
@@ -27,6 +28,7 @@ class Wallet:
     language: Language | None
     path: str | None
     index: int | None
+    w3: Web3 | None
 
     def __init__(self) -> None:
         """Create a new wallet.
@@ -44,41 +46,52 @@ def __init__(self) -> None:
         self.language = None
         self.path = ETHEREUM_DEFAULT_PATH
         self.index = Wallet.INDEX_DEFAULT
+        self.w3 = None
+
+
+    def nonce(self) -> int:
+        if not self.w3:
+            raise ValueError("Web3 instance not provided")
+
+        return self.w3.eth.get_transaction_count(self.address)
 
-    def transfer(
-        self, 
-        w3:Web3, 
-        to:str, 
-        amount:int,
-        gas:int|None=None, 
-        gas_price:int|None=None, 
-    ) -> str:
-        """Transfer a specified amount of wei to a specified address."""
-        if not self.account:
-            raise ValueError("Account not initialized")
-
-        if not gas:
-            gas = w3.eth.estimate_gas({
-                'to': to,
-                'from': self.address,
-                'value': amount
-            })
-
-        if not gas_price:
-            gas_price = w3.eth.gas_price
+
+    def balance(self) -> int:
+        if not self.w3:
+            raise ValueError("Web3 instance not provided")
+
+        return self.w3.eth.get_balance(self.address)
+
+
+    def transfer(self, to: Union [str, "Wallet"], amount: int, gas_price: int = None) -> str:
+        if not self.w3:
+            raise ValueError("Web3 instance not provided")
+        
+        if isinstance(to, Wallet):
+            to = to.address
+
+        nonce = self.nonce()
+        chain_id = self.w3.eth.chain_id
+        gas = 21000
+        gas_price = gas_price or self.w3.eth.gas_price
 
         tx = {
             "to": to,
             "value": amount,
-            "nonce": w3.eth.get_transaction_count(self.address),
+            "nonce": nonce,
+            "chainId": chain_id, # only replay-protected (EIP-155) transactions allowed over RPC
             "gas": gas,
             "gasPrice": gas_price,
         }
 
-        signed_tx = self.account.sign_transaction(tx)
-        tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
+        try:
+            signed = self.account.sign_transaction(tx)
+            tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction)
+
+        except Exception as e:
+            raise ValueError(f"Error sending transaction: {e}")
 
-        return tx_hash.hex()
+        return tx_hash.to_0x_hex()
 
 
     @classmethod
@@ -88,6 +101,7 @@ def create(
         language: Language = LANGUAGE_DEFAULT,
         index: int = INDEX_DEFAULT,
         password: str | None = None,
+        w3: Web3 | None = None,
         print_address: bool = True,  # noqa: FBT001, FBT002
     ) -> "Wallet":
         """Create a new wallet."""
@@ -116,6 +130,7 @@ def create(
 
         wallet.password = password
         wallet.vault = wallet.account.encrypt(password)  # type: ignore  # noqa: PGH003
+        wallet.w3 = w3
 
         return wallet
 
@@ -125,6 +140,7 @@ def from_mnemonic(
         index: int = INDEX_DEFAULT,
         password: str = generate_password(),
         path: str = ETHEREUM_DEFAULT_PATH,
+        w3: Web3 | None = None,
     ) -> "Wallet":
         """Create a new wallet from a provided mnemonic."""
         Wallet.validate_mnemonic(mnemonic)
@@ -134,6 +150,7 @@ def from_mnemonic(
         wallet.mnemonic = mnemonic
         wallet.index = index
         wallet.path = path
+        wallet.w3 = w3
 
         # modify path if index is provided
         if index: