diff --git a/example_eip7702.py b/example_eip7702.py new file mode 100644 index 0000000..2470eab --- /dev/null +++ b/example_eip7702.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +""" +EIP-7702 Delegation Example: Counter Contract + +This example demonstrates how to use EIP-7702 delegation utilities to sponsor +a delegated call to a Counter contract's increment method. + +Usage: + export SPONSOR_PRIVATE_KEY="0x..." + python example_eip7702.py + +The workflow: +1. Sponsor wallet (with existing funds) pays gas for the transaction +2. Delegate wallet (randomly created) authorizes setting its code to the Counter contract via EIP-7702 +3. Counter contract's increment method is executed on the delegate's account +4. The delegate's account code is set to the Counter contract during execution +""" + +import asyncio +import sys +import os +from typing import Annotated + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "packages/eth_rpc/src")) + +from eth_rpc import TransactionReceipt +from eth_rpc.contract import ProtocolBase, ContractFunc +from eth_rpc.delegation import sponsor_delegation +from eth_rpc.wallet import PrivateKeyWallet +from eth_rpc.types import METHOD, Name, NoArgs +from eth_typing import HexAddress, HexStr +from eth_rpc.networks import Sepolia + + +class Counter(ProtocolBase): + """ + Example Counter contract with increment method. + + Solidity equivalent: + contract Counter { + uint256 public count; + address public lastToEverUpdate; + + function increment() external { + count++; + lastToEverUpdate = msg.sender; + } + } + """ + increment: ContractFunc[NoArgs, None] = METHOD + number: ContractFunc[NoArgs, int] = METHOD + last_to_ever_update: Annotated[ + ContractFunc[NoArgs, HexAddress], + Name("lastToEverUpdate"), + ] = METHOD + + +async def main(): + """Demonstrate EIP-7702 delegation workflow with Counter contract""" + print("šŸ”— EIP-7702 Delegation Example: Counter Contract") + print("=" * 50) + + sponsor_private_key = os.getenv("SPONSOR_PRIVATE_KEY") + + if not sponsor_private_key: + print("āŒ Error: SPONSOR_PRIVATE_KEY environment variable not set") + print("Usage: export SPONSOR_PRIVATE_KEY='0x...' && python example_eip7702.py") + sys.exit(1) + + print("\n1. Setting up wallets...") + sponsor_wallet = PrivateKeyWallet[Sepolia](private_key=HexStr(sponsor_private_key)) + delegate_wallet = PrivateKeyWallet.create_new() + + print(f" Sponsor wallet: {sponsor_wallet.address} (has funds, pays gas fees)") + print(f" Delegate wallet: {delegate_wallet.address} (randomly created, authorizes code setting)") + + counter_address = HexAddress("0x0271297dcc0CceA3640bbaf34801025E6F63F448") + print(f" Counter contract: {counter_address}") + + print("\n2. Creating Counter contract instance...") + counter = Counter[Sepolia](address=counter_address) + print(f" Counter.increment function: {counter.increment}") + + print("\n3. Preparing increment call...") + increment_call_data = counter.increment().data + print(f" Increment call data: {increment_call_data}") + + # Create sponsored delegation transaction + print("\n4. Creating sponsored delegation transaction...") + print(" This transaction will:") + print(" - Be paid for by the sponsor wallet (which has ETH for gas)") + print(" - Set the delegate's account code to the Counter contract") + print(" - Execute the increment method within the same transaction") + print(" - Automatically handle network-aware nonce lookup") + + print(" Using the enhanced execute method with delegation...") + tx_hash = await counter.increment().execute( + wallet=sponsor_wallet, + delegate_wallet=delegate_wallet, + ) + print(f" āœ… Transaction sent using enhanced execute method: {tx_hash}") + + print(f"\n5. Demonstrating simple wallet delegation (without contract data)...") + print(" Using the wallet delegation utility method...") + + simple_delegate = PrivateKeyWallet.create_new() + print(f" New delegate wallet: {simple_delegate.address}") + + delegation_tx_hash = await simple_delegate.delegate_to_contract( + sponsor_wallet=sponsor_wallet, + contract_address=counter_address, + ) + print(f" Delegation transaction sent: {delegation_tx_hash}") + + print(f"\nšŸŽ‰ EIP-7702 delegation workflow complete!") + print(" Both utility methods successfully demonstrated.") + + print("Waiting for transaction to be mined...") + while True: + receipt = await TransactionReceipt[Sepolia].get_by_hash(tx_hash) + if receipt: + if receipt.status == 1: + print("Transaction mined successfully") + break + if receipt.status == 0: + raise Exception(f"Transaction failed: {receipt.status}") + await asyncio.sleep(4) + + counter = Counter[Sepolia](address=delegate_wallet.address) + print(f" Counter.number: {await counter.number().get()}") + print(f" Counter.last_to_ever_update: {await counter.last_to_ever_update().get()}") + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except Exception as e: + print(f"\nāŒ Example failed: {e}") + import traceback + traceback.print_exc() + sys.exit(1) diff --git a/packages/eth_rpc/src/eth_rpc/__init__.py b/packages/eth_rpc/src/eth_rpc/__init__.py index 6a9a105..f5e4da2 100644 --- a/packages/eth_rpc/src/eth_rpc/__init__.py +++ b/packages/eth_rpc/src/eth_rpc/__init__.py @@ -13,6 +13,7 @@ from .block import Block from .codegen import codegen from .contract import Contract, ContractFunc, EthResponse, FuncSignature, ProtocolBase +from .delegation import create_authorization_item, prepare_delegation_transaction, sponsor_delegation from .event import Event from .log import Log from .models import EventData @@ -47,12 +48,15 @@ "add_middleware", "codegen", "configure_rpc_from_env", + "create_authorization_item", "get_current_network", "get_selected_wallet", + "prepare_delegation_transaction", "set_alchemy_key", "set_default_network", - "set_selected_wallet", - "set_transport", "set_rpc_timeout", "set_rpc_url", + "set_selected_wallet", + "set_transport", + "sponsor_delegation", ] diff --git a/packages/eth_rpc/src/eth_rpc/contract/function.py b/packages/eth_rpc/src/eth_rpc/contract/function.py index c02e3e2..f37c9e2 100644 --- a/packages/eth_rpc/src/eth_rpc/contract/function.py +++ b/packages/eth_rpc/src/eth_rpc/contract/function.py @@ -21,6 +21,7 @@ from .._transport import _force_get_global_rpc from ..block import Block from ..constants import ADDRESS_ZERO +from ..delegation import sponsor_delegation from ..rpc.core import RPC from ..transaction import PreparedTransaction from ..utils import run @@ -530,8 +531,23 @@ async def _execute( max_fee_per_gas: Optional[int] = None, max_priority_fee_per_gas: Optional[int] = None, use_access_list: bool = False, + delegate_wallet: Optional["BaseWallet"] = None, + chain_id: Optional[int] = None, + gas: Optional[int] = None, sync: bool = True, ) -> HexStr: + # If delegate_wallet is provided, use sponsored delegation + if delegate_wallet is not None: + return await self._execute_sponsored( + sponsor_wallet=wallet, + delegate_wallet=delegate_wallet, + chain_id=chain_id, + nonce=nonce, + value=value, + gas=gas or 100000, + sync=sync, + ) + if sync is True: prepared_tx = self.prepare( wallet, @@ -575,6 +591,9 @@ def execute( max_fee_per_gas: Optional[int] = ..., max_priority_fee_per_gas: Optional[int] = ..., use_access_list: bool = ..., + delegate_wallet: Optional["BaseWallet"] = ..., + chain_id: Optional[int] = ..., + gas: Optional[int] = ..., ) -> HexStr: ... @overload @@ -588,6 +607,9 @@ def execute( max_fee_per_gas: Optional[int] = ..., max_priority_fee_per_gas: Optional[int] = ..., use_access_list: bool = ..., + delegate_wallet: Optional["BaseWallet"] = ..., + chain_id: Optional[int] = ..., + gas: Optional[int] = ..., ) -> Awaitable[HexStr]: ... def execute( @@ -600,6 +622,9 @@ def execute( max_fee_per_gas: Optional[int] = None, max_priority_fee_per_gas: Optional[int] = None, use_access_list: bool = False, + delegate_wallet: Optional["BaseWallet"] = None, + chain_id: Optional[int] = None, + gas: Optional[int] = None, sync: bool = False, ) -> MaybeAwaitable[HexStr]: return run( @@ -611,6 +636,88 @@ def execute( max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas, use_access_list=use_access_list, + delegate_wallet=delegate_wallet, + chain_id=chain_id, + gas=gas, + sync=sync, + ) + + async def _execute_sponsored( + self, + sponsor_wallet: "BaseWallet", + delegate_wallet: "BaseWallet", + *, + chain_id: Optional[int] = None, + nonce: Optional[int] = None, + value: int = 0, + gas: int = 100000, + sync: bool = False, + ) -> HexStr: + sponsored_tx = await sponsor_delegation( + sponsor_wallet=sponsor_wallet, + delegate_wallet=delegate_wallet, + contract_address=self.address, + chain_id=chain_id, + nonce=nonce, + value=value, + data=self.data, + gas=gas, + ) + signed_tx = sponsor_wallet.sign_transaction(sponsored_tx) + if sync: + return ( + sponsor_wallet[self._network] + .send_raw_transaction(HexStr("0x" + signed_tx.raw_transaction)) + .sync + ) + return await sponsor_wallet[self._network].send_raw_transaction( + HexStr("0x" + signed_tx.raw_transaction) + ) + + @overload + def execute_sponsored( + self, + sponsor_wallet: "BaseWallet", + delegate_wallet: "BaseWallet", + *, + sync: Literal[True], + chain_id: Optional[int] = ..., + nonce: Optional[int] = ..., + value: int = ..., + gas: int = ..., + ) -> HexStr: ... + + @overload + def execute_sponsored( + self, + sponsor_wallet: "BaseWallet", + delegate_wallet: "BaseWallet", + *, + chain_id: Optional[int] = ..., + nonce: Optional[int] = ..., + value: int = ..., + gas: int = ..., + ) -> Awaitable[HexStr]: ... + + def execute_sponsored( + self, + sponsor_wallet: "BaseWallet", + delegate_wallet: "BaseWallet", + *, + chain_id: Optional[int] = None, + nonce: Optional[int] = None, + value: int = 0, + gas: int = 100000, + sync: bool = False, + ) -> MaybeAwaitable[HexStr]: + return run( + self._execute_sponsored, + sponsor_wallet, + delegate_wallet, + chain_id=chain_id, + nonce=nonce, + value=value, + gas=gas, sync=sync, ) @@ -710,6 +817,9 @@ def execute( # type: ignore max_fee_per_gas: Optional[int] = None, max_priority_fee_per_gas: Optional[int] = None, use_access_list: bool = False, + delegate_wallet: Optional[BaseWallet] = None, + chain_id: Optional[int] = None, + gas: Optional[int] = None, ) -> HexStr: return super().execute( wallet, @@ -718,5 +828,28 @@ def execute( # type: ignore max_fee_per_gas=max_fee_per_gas, max_priority_fee_per_gas=max_priority_fee_per_gas, use_access_list=use_access_list, + delegate_wallet=delegate_wallet, + chain_id=chain_id, + gas=gas, + sync=self.SYNC, + ) + + def execute_sponsored( # type: ignore + self, + sponsor_wallet: BaseWallet, + delegate_wallet: BaseWallet, + *, + chain_id: Optional[int] = None, + nonce: Optional[int] = None, + value: int = 0, + gas: int = 100000, + ) -> HexStr: + return super().execute_sponsored( + sponsor_wallet, + delegate_wallet, + chain_id=chain_id, + nonce=nonce, + value=value, + gas=gas, sync=self.SYNC, ) diff --git a/packages/eth_rpc/src/eth_rpc/delegation.py b/packages/eth_rpc/src/eth_rpc/delegation.py new file mode 100644 index 0000000..0055c7e --- /dev/null +++ b/packages/eth_rpc/src/eth_rpc/delegation.py @@ -0,0 +1,155 @@ +from typing import Optional + +from eth_typing import HexAddress, HexStr +from eth_account import Account as EthAccount + +from .types import HexInteger +from .types.transaction import AuthorizationItem +from .transaction import PreparedTransaction +from .wallet import BaseWallet + + +def create_authorization_item( + chain_id: int, + contract_address: HexAddress, + nonce: int, + private_key: HexStr, +) -> AuthorizationItem: + """ + Create a signed authorization item for EIP-7702 delegation. + + Args: + chain_id: The chain ID for the authorization + contract_address: The contract address to set as the EOA's code + nonce: The nonce for this authorization + private_key: Private key of the EOA to sign the authorization + + Returns: + Signed AuthorizationItem + """ + auth = { + "chainId": chain_id, + "address": contract_address, + "nonce": nonce, + } + + account = EthAccount.from_key(private_key) + signed_auth = account.sign_authorization(auth) + + return AuthorizationItem( + chain_id=HexInteger(chain_id), + address=contract_address, + nonce=HexInteger(nonce), + y_parity=HexInteger(signed_auth.y_parity), + r=HexStr(hex(signed_auth.r)), + s=HexStr(hex(signed_auth.s)), + ) + + +async def prepare_delegation_transaction( + wallet: BaseWallet, + to: HexAddress, + authorization_list: list[AuthorizationItem], + value: int = 0, + data: HexStr = HexStr("0x"), + gas: int = 100000, + max_fee_per_gas: int = 20000000000, + max_priority_fee_per_gas: int = 1000000000, + nonce: Optional[int] = None, +) -> PreparedTransaction: + """ + Prepare an EIP-7702 delegation transaction. + + Args: + wallet: Wallet to send the transaction from (sponsor) + to: Target address for the transaction (delegate's EOA address) + authorization_list: List of authorization items for setting EOA code + value: ETH value to send (default: 0) + data: Transaction data for contract function calls (default: "0x") + gas: Gas limit (auto-estimated if None) + max_fee_per_gas: Maximum fee per gas + max_priority_fee_per_gas: Maximum priority fee per gas + nonce: Transaction nonce (auto-detected from wallet if None) + + Returns: + PreparedTransaction with type 4 and authorization list + """ + from .types import HexInteger + + if nonce is None: + nonce = await wallet[wallet._network].get_nonce() + + base_tx = PreparedTransaction( + data=data, + to=to, + gas=HexInteger(gas), + max_fee_per_gas=max_fee_per_gas, + max_priority_fee_per_gas=max_priority_fee_per_gas, + nonce=nonce, + value=value, + authorization_list=authorization_list, + chain_id=authorization_list[0].chain_id, + type=4, + ) + + return base_tx + + +async def sponsor_delegation( + sponsor_wallet: BaseWallet, + delegate_wallet: BaseWallet, + contract_address: HexAddress, + chain_id: Optional[int] = None, + nonce: Optional[int] = None, + value: int = 0, + data: HexStr = HexStr("0x"), + gas: int = 100000, +) -> PreparedTransaction: + """ + Create a sponsored delegation transaction where the sponsor pays gas + for setting the delegate's code to a contract and executing contract functions. + + Automatically handles network-aware nonce lookup and sponsor == delegate cases. + + Args: + sponsor_wallet: Wallet that will pay for gas + delegate_wallet: Wallet that will have its code set to the contract + contract_address: Contract address to set as the delegate's code + chain_id: Chain ID for the authorization (auto-detected from sponsor wallet if None) + nonce: Delegate's current nonce for authorization (auto-detected if None) + value: ETH value to send + data: Transaction data (contract function calls) + gas: Gas limit for the transaction (default: 100000) + + Returns: + PreparedTransaction ready to be signed by sponsor + """ + if chain_id is None: + rpc = sponsor_wallet._rpc() + chain_id = rpc.network.chain_id + + if nonce is None: + nonce = await delegate_wallet[sponsor_wallet._network].get_nonce() + + sponsor_is_delegate = sponsor_wallet.address == delegate_wallet.address + if sponsor_is_delegate: + auth_nonce = nonce + 1 + else: + auth_nonce = nonce + + auth_item = create_authorization_item( + chain_id=chain_id, + contract_address=contract_address, + nonce=auth_nonce, + private_key=delegate_wallet.private_key, + ) + + return await prepare_delegation_transaction( + wallet=sponsor_wallet, + to=delegate_wallet.address, + authorization_list=[auth_item], + value=value, + data=data, + gas=gas, + nonce=nonce if sponsor_is_delegate else None, + ) diff --git a/packages/eth_rpc/src/eth_rpc/models/transaction.py b/packages/eth_rpc/src/eth_rpc/models/transaction.py index e9eba1a..098f846 100644 --- a/packages/eth_rpc/src/eth_rpc/models/transaction.py +++ b/packages/eth_rpc/src/eth_rpc/models/transaction.py @@ -1,6 +1,7 @@ from typing import Optional from eth_rpc.types import HexInteger +from eth_rpc.types.transaction import AuthorizationItem from eth_typing import HexStr from pydantic import BaseModel, ConfigDict, Field from pydantic.alias_generators import to_camel @@ -17,6 +18,7 @@ class BaseTransaction(BaseModel): hash: HexStr access_list: Optional[list[AccessList]] = None + authorization_list: Optional[list[AuthorizationItem]] = None chain_id: Optional[HexInteger] = None from_: HexStr = Field(alias="from") gas: HexInteger diff --git a/packages/eth_rpc/src/eth_rpc/transaction.py b/packages/eth_rpc/src/eth_rpc/transaction.py index f6493a3..98bd9fb 100644 --- a/packages/eth_rpc/src/eth_rpc/transaction.py +++ b/packages/eth_rpc/src/eth_rpc/transaction.py @@ -7,6 +7,7 @@ from eth_rpc.models import AccessList, PendingTransaction from eth_rpc.models import Transaction as TransactionModel from eth_rpc.models import TransactionReceipt as TransactionReceiptModel +from eth_rpc.types.transaction import AuthorizationItem from eth_rpc.types.args import ( GetTransactionByBlockHash, GetTransactionByBlockNumber, @@ -55,6 +56,7 @@ class PreparedTransaction(BaseModel): to: HexAddress value: int access_list: Optional[list[AccessList]] = None + authorization_list: Optional[list[AuthorizationItem]] = None chain_id: int def model_dump(self, *args, exclude_none=True, by_alias=True, **kwargs): @@ -64,7 +66,9 @@ def model_dump(self, *args, exclude_none=True, by_alias=True, **kwargs): @model_validator(mode="after") def validate_xor(self): - if self.max_fee_per_gas is not None: + if self.authorization_list is not None: + self.type = 4 + elif self.max_fee_per_gas is not None: self.type = 2 else: self.type = 1 @@ -87,6 +91,44 @@ def get_by_hash( ), ) + @classmethod + async def wait_until_finalized( + cls, + tx_hash: HexStr, + sleep_time: float = 4.0, + timeout: Optional[float] = None, + ) -> "TransactionReceipt[Network]": + """ + Wait until a transaction is finalized by polling for its receipt. + + Args: + tx_hash: The transaction hash to wait for + sleep_time: Time to sleep between polling attempts (default: 4.0 seconds) + timeout: Maximum time to wait before giving up (default: None for no timeout) + + Returns: + The finalized transaction receipt + + Raises: + Exception: If the transaction fails (status == 0) + asyncio.TimeoutError: If timeout is reached before finalization + """ + import time + start_time = time.time() if timeout else None + + while True: + receipt = await cls.get_by_hash(tx_hash) + if receipt: + if receipt.status == 1: + return receipt + elif receipt.status == 0: + raise Exception(f"Transaction failed: {receipt.status}") + + if timeout and start_time and (time.time() - start_time) > timeout: + raise asyncio.TimeoutError(f"Transaction {tx_hash} not finalized within {timeout} seconds") + + await asyncio.sleep(sleep_time) + @classmethod def get_block_receipts( self, diff --git a/packages/eth_rpc/src/eth_rpc/types/transaction.py b/packages/eth_rpc/src/eth_rpc/types/transaction.py index c6a91ae..7fa1206 100644 --- a/packages/eth_rpc/src/eth_rpc/types/transaction.py +++ b/packages/eth_rpc/src/eth_rpc/types/transaction.py @@ -1,5 +1,23 @@ -from eth_typing import HexStr -from pydantic import BaseModel +from eth_typing import HexAddress, HexStr +from pydantic import BaseModel, ConfigDict +from pydantic.alias_generators import to_camel + +from . import HexInteger + + +class AuthorizationItem(BaseModel): + model_config = ConfigDict( + alias_generator=to_camel, + populate_by_name=True, + from_attributes=True, + ) + + chain_id: HexInteger + address: HexAddress + nonce: HexInteger + y_parity: HexInteger + r: HexStr + s: HexStr class SignedTransaction(BaseModel): diff --git a/packages/eth_rpc/src/eth_rpc/utils/__init__.py b/packages/eth_rpc/src/eth_rpc/utils/__init__.py index 638b698..4ca7a7b 100644 --- a/packages/eth_rpc/src/eth_rpc/utils/__init__.py +++ b/packages/eth_rpc/src/eth_rpc/utils/__init__.py @@ -2,6 +2,13 @@ from .bloom import BloomFilter from .datetime import convert_datetime_to_iso_8601, load_datetime_string from .dual_async import handle_maybe_awaitable, run +from .event_receipt import ( + EventReceiptUtility, + get_events_from_receipt, + get_events_from_tx_hash, + get_single_event_from_receipt, + get_single_event_from_tx_hash, +) from .model import RPCModel from .streams import acombine, combine, ordered_iterator, sort_key from .types import is_annotation, to_bytes32, to_hex_str, to_topic, transform_primitive @@ -15,6 +22,11 @@ "to_bytes32", "combine", "convert_datetime_to_iso_8601", + "EventReceiptUtility", + "get_events_from_receipt", + "get_events_from_tx_hash", + "get_single_event_from_receipt", + "get_single_event_from_tx_hash", "handle_maybe_awaitable", "is_annotation", "load_datetime_string", diff --git a/packages/eth_rpc/src/eth_rpc/utils/event_receipt.py b/packages/eth_rpc/src/eth_rpc/utils/event_receipt.py new file mode 100644 index 0000000..abfcae6 --- /dev/null +++ b/packages/eth_rpc/src/eth_rpc/utils/event_receipt.py @@ -0,0 +1,138 @@ +from typing import TYPE_CHECKING, Optional, TypeVar + +from eth_typing import HexStr + +if TYPE_CHECKING: + from ..event import Event + from ..models import EventData, TransactionReceipt + from ..transaction import Transaction + +T = TypeVar("T") + + +class EventReceiptUtility: + """ + Utility for extracting and decoding events from transaction receipts. + + This utility takes Event types and transaction receipts/hashes, matches events + by their topic0 signatures, and returns properly decoded EventData objects. + """ + + @staticmethod + async def get_events_from_receipt( + events: list["Event[T]"], receipt: "TransactionReceipt" + ) -> list["EventData[T]"]: + """ + Extract and decode events from a transaction receipt. + + Args: + events: List of Event types to match against + receipt: Transaction receipt containing logs + + Returns: + List of decoded EventData objects for matched events + """ + matched_events = [] + + for log in receipt.logs: + for event in events: + if event.match(log): + try: + event_data = event.process_log(log) + matched_events.append(event_data) + except Exception: + continue + + return matched_events + + @staticmethod + async def get_events_from_tx_hash( + events: list["Event[T]"], tx_hash: HexStr + ) -> list["EventData[T]"]: + """ + Extract and decode events from a transaction hash. + + Args: + events: List of Event types to match against + tx_hash: Transaction hash to get receipt for + + Returns: + List of decoded EventData objects for matched events + """ + from ..transaction import Transaction + + receipt = await Transaction.get_receipt_by_hash(tx_hash) + if not receipt: + return [] + + return await EventReceiptUtility.get_events_from_receipt(events, receipt) + + @staticmethod + async def get_single_event_from_receipt( + event: "Event[T]", receipt: "TransactionReceipt" + ) -> Optional["EventData[T]"]: + """ + Extract and decode a single event type from a transaction receipt. + + Args: + event: Event type to match against + receipt: Transaction receipt containing logs + + Returns: + First matching decoded EventData object, or None if no matches + """ + results = await EventReceiptUtility.get_events_from_receipt([event], receipt) + return results[0] if results else None + + @staticmethod + async def get_single_event_from_tx_hash( + event: "Event[T]", tx_hash: HexStr + ) -> Optional["EventData[T]"]: + """ + Extract and decode a single event type from a transaction hash. + + Args: + event: Event type to match against + tx_hash: Transaction hash to get receipt for + + Returns: + First matching decoded EventData object, or None if no matches + """ + results = await EventReceiptUtility.get_events_from_tx_hash([event], tx_hash) + return results[0] if results else None + + +async def get_events_from_receipt( + events: list["Event[T]"], receipt: "TransactionReceipt" +) -> list["EventData[T]"]: + """ + Convenience function to extract and decode events from a transaction receipt. + """ + return await EventReceiptUtility.get_events_from_receipt(events, receipt) + + +async def get_events_from_tx_hash( + events: list["Event[T]"], tx_hash: HexStr +) -> list["EventData[T]"]: + """ + Convenience function to extract and decode events from a transaction hash. + """ + return await EventReceiptUtility.get_events_from_tx_hash(events, tx_hash) + + +async def get_single_event_from_receipt( + event: "Event[T]", receipt: "TransactionReceipt" +) -> Optional["EventData[T]"]: + """ + Convenience function to extract and decode a single event from a transaction receipt. + """ + return await EventReceiptUtility.get_single_event_from_receipt(event, receipt) + + +async def get_single_event_from_tx_hash( + event: "Event[T]", tx_hash: HexStr +) -> Optional["EventData[T]"]: + """ + Convenience function to extract and decode a single event from a transaction hash. + """ + return await EventReceiptUtility.get_single_event_from_tx_hash(event, tx_hash) diff --git a/packages/eth_rpc/src/eth_rpc/wallet.py b/packages/eth_rpc/src/eth_rpc/wallet.py index 46a2def..3bceb3f 100644 --- a/packages/eth_rpc/src/eth_rpc/wallet.py +++ b/packages/eth_rpc/src/eth_rpc/wallet.py @@ -48,6 +48,48 @@ def send_raw_transaction( self, tx: HexStr ) -> RPCResponseModel[RawTransaction, HexStr]: ... + async def delegate_to_contract( + self, + sponsor_wallet: "BaseWallet", + contract_address: HexAddress, + *, + chain_id: Optional[int] = None, + nonce: Optional[int] = None, + value: int = 0, + gas: int = 100000, + ) -> HexStr: + """ + Simple delegation method for setting this wallet's code to a contract. + + Args: + sponsor_wallet: Wallet that will pay for gas + contract_address: Contract address to set as this wallet's code + chain_id: Chain ID for the authorization (auto-detected if None) + nonce: This wallet's current nonce for authorization (auto-detected if None) + value: ETH value to send + gas: Gas limit for the transaction + + Returns: + Transaction hash + """ + from .delegation import sponsor_delegation + + sponsored_tx = await sponsor_delegation( + sponsor_wallet=sponsor_wallet, + delegate_wallet=self, + contract_address=contract_address, + chain_id=chain_id, + nonce=nonce, + value=value, + data=HexStr("0x"), + gas=gas, + ) + signed_tx = sponsor_wallet.sign_transaction(sponsored_tx) + result = await sponsor_wallet[sponsor_wallet._network].send_raw_transaction( + HexStr("0x" + signed_tx.raw_transaction) + ) + return result + class MockWallet(BaseWallet): _address: HexAddress = PrivateAttr() diff --git a/packages/eth_rpc/tests/test_event_receipt.py b/packages/eth_rpc/tests/test_event_receipt.py new file mode 100644 index 0000000..da217a6 --- /dev/null +++ b/packages/eth_rpc/tests/test_event_receipt.py @@ -0,0 +1,340 @@ +from typing import Annotated + +import pytest +from eth_rpc import Event +from eth_rpc.models import Log, TransactionReceipt +from eth_rpc.types import Indexed, primitives +from eth_rpc.utils.event_receipt import ( + EventReceiptUtility, + get_events_from_receipt, + get_events_from_tx_hash, + get_single_event_from_receipt, + get_single_event_from_tx_hash, +) +from pydantic import BaseModel + + +class TransferEventType(BaseModel): + sender: Annotated[primitives.address, Indexed] + recipient: Annotated[primitives.address, Indexed] + amount: primitives.uint256 + + +class ApprovalEventType(BaseModel): + owner: Annotated[primitives.address, Indexed] + spender: Annotated[primitives.address, Indexed] + value: primitives.uint256 + + +TransferEvent = Event[TransferEventType](name="Transfer") +ApprovalEvent = Event[ApprovalEventType](name="Approval") +from eth_typing import HexStr + +TRANSFER_TX_HASH = HexStr( + "0xc1b74e10a88ad2bc610432a182ae6d8200bd684704c44dfd0b915b86d4554211" +) +APPROVAL_TX_HASH = HexStr( + "0x353bbe9c19982849849227f8745a7fb633502bbabde3068f2aeb3295083bc78e" +) + + +@pytest.fixture(scope="session") +def transfer_receipt(): + """Construct real transaction receipt with Transfer events from raw data.""" + logs_data = [ + { + "transaction_hash": "0xc1b74e10a88ad2bc610432a182ae6d8200bd684704c44dfd0b915b86d4554211", + "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", + "block_hash": "0xa5542fac95d9d23fa3481614579810c6314931cbb3082b4c241781d8eb7650a3", + "block_number": "0x1601e91", + "data": "0x0000000000000000000000000000000000000000000000000000000011898e69", + "log_index": "0x148", + "removed": False, + "topics": [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x000000000000000000000000e0554a476a092703abdb3ef35c80e0d76d32939f", + "0x0000000000000000000000001111111254eeb25477b68fb85ed929f73a960582", + ], + "transaction_index": "0x9d", + }, + { + "transaction_hash": "0xc1b74e10a88ad2bc610432a182ae6d8200bd684704c44dfd0b915b86d4554211", + "address": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2", + "block_hash": "0xa5542fac95d9d23fa3481614579810c6314931cbb3082b4c241781d8eb7650a3", + "block_number": "0x1601e91", + "data": "0x0000000000000000000000000000000000000000000000000121de58aa468493", + "log_index": "0x149", + "removed": False, + "topics": [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x0000000000000000000000005141b82f5ffda4c6fe1e372978f1c5427640a190", + "0x000000000000000000000000e0554a476a092703abdb3ef35c80e0d76d32939f", + ], + "transaction_index": "0x9d", + }, + { + "transaction_hash": "0xc1b74e10a88ad2bc610432a182ae6d8200bd684704c44dfd0b915b86d4554211", + "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", + "block_hash": "0xa5542fac95d9d23fa3481614579810c6314931cbb3082b4c241781d8eb7650a3", + "block_number": "0x1601e91", + "data": "0x0000000000000000000000000000000000000000000000000000000011898e69", + "log_index": "0x14b", + "removed": False, + "topics": [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x0000000000000000000000001111111254eeb25477b68fb85ed929f73a960582", + "0x000000000000000000000000f621fb08bbe51af70e7e0f4ea63496894166ff7f", + ], + "transaction_index": "0x9d", + }, + { + "transaction_hash": "0xc1b74e10a88ad2bc610432a182ae6d8200bd684704c44dfd0b915b86d4554211", + "address": "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48", + "block_hash": "0xa5542fac95d9d23fa3481614579810c6314931cbb3082b4c241781d8eb7650a3", + "block_number": "0x1601e91", + "data": "0x0000000000000000000000000000000000000000000000000000000011898e69", + "log_index": "0x14c", + "removed": False, + "topics": [ + "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef", + "0x000000000000000000000000f621fb08bbe51af70e7e0f4ea63496894166ff7f", + "0x000000000000000000000000b8f275fbf7a959f4bce59999a2ef122a099e81a8", + ], + "transaction_index": "0x9d", + }, + ] + + logs = [Log(**log_data) for log_data in logs_data] + + return TransactionReceipt( + transaction_hash=TRANSFER_TX_HASH, + block_hash="0xa5542fac95d9d23fa3481614579810c6314931cbb3082b4c241781d8eb7650a3", + block_number="0x1601e91", + logs=logs, + contract_address=None, + effective_gas_price="0x360bc194", + cumulative_gas_used="0x112d0aa", + from_="0x2b2cf32c66ea5d1daf9ec70daf7146f445719664", + gas_used="0x7a251", + logs_bloom="0x200000000200000000000001001008000040000000004000000000008000000001000000000000000000000020a400000080020000000000000000080000100000000080808000008000100000000000000001000100000008002000004200000820000000012000010000800000000020000000000000010000810000000004000000000000000000000080080000201010000000020801050000080000000000000280000000000080004020000000400000000800080802000000000001002000400000000001000000000040000000000000000004000000028200000200000800000008400800480080001000000000000400000000000000000", + status="0x1", + to="0x1231deb6f5749ef6ce6943a275a1d3e7486f4eae", + transaction_index="0x9d", + type="0x2", + ) + + +@pytest.fixture(scope="session") +def approval_receipt(): + """Construct real transaction receipt with Approval events from raw data.""" + logs_data = [ + { + "transaction_hash": "0x353bbe9c19982849849227f8745a7fb633502bbabde3068f2aeb3295083bc78e", + "address": "0x30a538effd91acefb1b12ce9bc0074ed18c9dfc9", + "block_hash": "0x5f42f60b97c076b40d8cad6b8f6cb1dc8f08f50ce0bc6e429e37d2541c96a22d", + "block_number": "0x15c604ab", + "data": "0x0000000000000000000000000000000000000000000000000000865666fd1589", + "log_index": "0x5", + "removed": False, + "topics": [ + "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925", + "0x000000000000000000000000353801845f8ae3ece792f5597cd96a17c46e6def", + "0x00000000000000000000000070cbb871e8f30fc8ce23609e9e0ea87b6b222f58", + ], + "transaction_index": "0x3", + } + ] + + logs = [Log(**log_data) for log_data in logs_data] + + return TransactionReceipt( + transaction_hash=APPROVAL_TX_HASH, + block_hash="0x5f42f60b97c076b40d8cad6b8f6cb1dc8f08f50ce0bc6e429e37d2541c96a22d", + block_number="0x15c604ab", + logs=logs, + contract_address=None, + effective_gas_price="0x989680", + cumulative_gas_used="0x6b34d", + from_="0x353801845f8ae3ece792f5597cd96a17c46e6def", + gas_used="0x7844", + logs_bloom="0x800000000000000000000000000000000000000000080000000000000000000004000000000000000000000000000000000000000000000000000200000000000000000000000000000400000000000000000000000000000000000000000000000000000000000400000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000001000000000000000000000000000000000000000000000000000000000010010000000000000000000000010000000000000000000000000000000000000000000000000000000000000", + status="0x1", + to="0x30a538effd91acefb1b12ce9bc0074ed18c9dfc9", + transaction_index="0x3", + type="0x2", + ) + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_receipt_transfer_events(transfer_receipt): + """Test extracting Transfer events from constructed transaction receipt.""" + events = [TransferEvent] + result = await EventReceiptUtility.get_events_from_receipt(events, transfer_receipt) + + # Should find 4 Transfer events in this transaction + assert len(result) == 4 + assert all(event_data.name == "Transfer" for event_data in result) + assert all(hasattr(event_data.event, "sender") for event_data in result) + assert all(hasattr(event_data.event, "recipient") for event_data in result) + assert all(hasattr(event_data.event, "amount") for event_data in result) + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_receipt_approval_events(approval_receipt): + """Test extracting Approval events from constructed transaction receipt.""" + events = [ApprovalEvent] + result = await EventReceiptUtility.get_events_from_receipt(events, approval_receipt) + + # Should find 1 Approval event in this transaction + assert len(result) == 1 + assert result[0].name == "Approval" + assert hasattr(result[0].event, "owner") + assert hasattr(result[0].event, "spender") + assert hasattr(result[0].event, "value") + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_receipt_no_matches(transfer_receipt): + """Test extracting events with no matches.""" + events = [ApprovalEvent] # Looking for Approval in Transfer transaction + result = await EventReceiptUtility.get_events_from_receipt(events, transfer_receipt) + + assert len(result) == 0 + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_receipt_multiple_event_types(transfer_receipt): + """Test extracting multiple event types from receipt.""" + events = [TransferEvent, ApprovalEvent] + result = await EventReceiptUtility.get_events_from_receipt(events, transfer_receipt) + + # Should only find Transfer events in this transaction + assert len(result) == 4 + assert all(event_data.name == "Transfer" for event_data in result) + + +@pytest.mark.asyncio(scope="session") +async def test_get_single_event_from_receipt(transfer_receipt): + """Test extracting single event from receipt.""" + result = await EventReceiptUtility.get_single_event_from_receipt( + TransferEvent, transfer_receipt + ) + + assert result is not None + assert result.name == "Transfer" + + +@pytest.mark.asyncio(scope="session") +async def test_get_single_event_from_receipt_no_match(transfer_receipt): + """Test extracting single event with no matches.""" + result = await EventReceiptUtility.get_single_event_from_receipt( + ApprovalEvent, transfer_receipt + ) + + assert result is None + + +@pytest.mark.asyncio(scope="session") +async def test_get_single_event_from_receipt_approval(approval_receipt): + """Test extracting single Approval event from receipt.""" + result = await EventReceiptUtility.get_single_event_from_receipt( + ApprovalEvent, approval_receipt + ) + + assert result is not None + assert result.name == "Approval" + + +@pytest.mark.asyncio(scope="session") +async def test_convenience_functions(transfer_receipt): + """Test that convenience functions work correctly.""" + result1 = await get_events_from_receipt([TransferEvent], transfer_receipt) + assert len(result1) == 4 + + result2 = await get_single_event_from_receipt(TransferEvent, transfer_receipt) + assert result2 is not None + + +@pytest.mark.asyncio(scope="session") +async def test_convenience_functions_with_approval(approval_receipt): + """Test convenience functions with Approval events.""" + result1 = await get_events_from_receipt([ApprovalEvent], approval_receipt) + assert len(result1) == 1 + + result2 = await get_single_event_from_receipt(ApprovalEvent, approval_receipt) + assert result2 is not None + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_tx_hash_mocked(transfer_receipt, monkeypatch): + """Test extracting events from transaction hash with mocked RPC call.""" + from eth_rpc.transaction import Transaction + + async def mock_get_receipt_by_hash(tx_hash): + return transfer_receipt + + monkeypatch.setattr(Transaction, "get_receipt_by_hash", mock_get_receipt_by_hash) + + events = [TransferEvent] + result = await EventReceiptUtility.get_events_from_tx_hash(events, TRANSFER_TX_HASH) + + assert len(result) == 4 + assert all(event_data.name == "Transfer" for event_data in result) + + +@pytest.mark.asyncio(scope="session") +async def test_get_single_event_from_tx_hash_mocked(approval_receipt, monkeypatch): + """Test extracting single event from transaction hash with mocked RPC call.""" + from eth_rpc.transaction import Transaction + + async def mock_get_receipt_by_hash(tx_hash): + return approval_receipt + + monkeypatch.setattr(Transaction, "get_receipt_by_hash", mock_get_receipt_by_hash) + + result = await EventReceiptUtility.get_single_event_from_tx_hash( + ApprovalEvent, APPROVAL_TX_HASH + ) + + assert result is not None + assert result.name == "Approval" + + +@pytest.mark.asyncio(scope="session") +async def test_get_events_from_tx_hash_invalid_hash_mocked(monkeypatch): + """Test handling of invalid transaction hash with mocked RPC call.""" + from eth_rpc.transaction import Transaction + + async def mock_get_receipt_by_hash(tx_hash): + return None + + monkeypatch.setattr(Transaction, "get_receipt_by_hash", mock_get_receipt_by_hash) + + invalid_hash = HexStr("0x" + "0" * 64) + events = [TransferEvent] + + result = await EventReceiptUtility.get_events_from_tx_hash(events, invalid_hash) + assert len(result) == 0 + + +@pytest.mark.asyncio(scope="session") +async def test_convenience_functions_from_tx_hash_mocked( + transfer_receipt, approval_receipt, monkeypatch +): + """Test convenience functions with transaction hash and mocked RPC calls.""" + from eth_rpc.transaction import Transaction + + async def mock_get_receipt_by_hash(tx_hash): + if tx_hash == TRANSFER_TX_HASH: + return transfer_receipt + elif tx_hash == APPROVAL_TX_HASH: + return approval_receipt + return None + + monkeypatch.setattr(Transaction, "get_receipt_by_hash", mock_get_receipt_by_hash) + + result1 = await get_events_from_tx_hash([TransferEvent], TRANSFER_TX_HASH) + assert len(result1) == 4 + + result2 = await get_single_event_from_tx_hash(ApprovalEvent, APPROVAL_TX_HASH) + assert result2 is not None