Adding typing to process_receipts via process_receipts_typed. Ver…

…sion bump to v0.0.41 (#133) Major updates: - Adding `process_receipts_typed` that returns a dataclass typed event when given a transaction receipt. - Adding `combomethod_typed` that allows for combomethods that preserve types, and propagating the change throughout. - Similar to solution in eth-utils here: ethereum/eth-utils#264 - Renaming `get_typed_logs` to `get_logs_typed` for consistency. - Adding `BaseEventArgs` and frozen dataclass attribute for arg subclassing. This also ensures the `arg` field exists in all event types.
delvtech · Oct 7, 2024 · 512e82e · 512e82e
1 parent d696dca
commit 512e82e
Show file tree

Hide file tree

Showing 47 changed files with 258 additions and 440 deletions.
diff --git a/example/types/ExampleContract.py b/example/types/ExampleContract.py
@@ -1,6 +1,6 @@
 """A web3.py Contract class for the Example contract.
 
-DO NOT EDIT.  This file was generated by pypechain v0.0.40.
+DO NOT EDIT.  This file was generated by pypechain v0.0.41.
 See documentation at https://github.com/delvtech/pypechain """
 
 # contracts have PascalCase names
@@ -35,6 +35,7 @@
 from hexbytes import HexBytes
 from typing_extensions import Self
 from web3 import Web3
+from web3._utils.events import EventLogErrorFlags
 from web3._utils.filters import LogFilter
 from web3.contract.contract import (
     Contract,
@@ -44,9 +45,10 @@
     ContractFunction,
     ContractFunctions,
 )
-from web3.types import BlockIdentifier, StateOverride, TxParams
+from web3.logs import WARN
+from web3.types import BlockIdentifier, StateOverride, TxParams, TxReceipt
 
-from pypechain.core import dataclass_to_tuple, get_abi_input_types, rename_returned_types
+from pypechain.core import combomethod_typed, dataclass_to_tuple, get_abi_input_types, rename_returned_types
 
 from .ExampleTypes import FlipEvent, FlopEvent, InnerStruct, NestedStruct, SimpleStruct
 
@@ -474,16 +476,13 @@ class ExampleFlipContractEvent(ContractEvent):
     # super() get_logs and create_filter methods are generic, while our version adds values & types
     # pylint: disable=arguments-differ
 
-    # @combomethod destroys return types, so we are redefining functions as both class and instance
-    # pylint: disable=function-redefined
-
     # pylint: disable=useless-parent-delegation
     def __init__(self, *argument_names: tuple[str]) -> None:
         super().__init__(*argument_names)
 
-    # We ignore types here for function redefinition
-    def get_typed_logs(  # type: ignore
-        self: "ExampleFlipContractEvent",
+    @combomethod_typed
+    def get_logs_typed(
+        self,
         argument_filters: dict[str, Any] | None = None,
         from_block: BlockIdentifier | None = None,
         to_block: BlockIdentifier | None = None,
@@ -509,19 +508,10 @@ def get_typed_logs(  # type: ignore
             for abi_event in abi_events
         ]
 
-    @classmethod
-    # We ignore types here for function redefinition
-    def get_typed_logs(  # type: ignore
-        cls: Type["ExampleFlipContractEvent"],
-        argument_filters: dict[str, Any] | None = None,
-        from_block: BlockIdentifier | None = None,
-        to_block: BlockIdentifier | None = None,
-        block_hash: HexBytes | None = None,
-    ) -> Iterable[FlipEvent]:
-        """Extension of `get_logs` that return a typed dataclass of the event."""
-        abi_events = super().get_logs(
-            argument_filters=argument_filters, from_block=from_block, to_block=to_block, block_hash=block_hash
-        )
+    @combomethod_typed
+    def process_receipt_typed(self, txn_receipt: TxReceipt, errors: EventLogErrorFlags = WARN) -> Iterable[FlipEvent]:
+        """Extension of `process_receipt` that return a typed dataclass of the event."""
+        abi_events = super().process_receipt(txn_receipt, errors)
         # TODO there may be issues with this function if the user uses a middleware that changes event structure.
         return [
             FlipEvent(
@@ -538,29 +528,9 @@ def get_typed_logs(  # type: ignore
             for abi_event in abi_events
         ]
 
+    @combomethod_typed
     def create_filter(  # type: ignore
-        self: "ExampleFlipContractEvent",
-        *,  # PEP 3102
-        argument_filters: dict[str, Any] | None = None,
-        from_block: BlockIdentifier | None = None,
-        to_block: BlockIdentifier = "latest",
-        address: ChecksumAddress | None = None,
-        topics: Sequence[Any] | None = None,
-    ) -> LogFilter:
-        return cast(
-            LogFilter,
-            super().create_filter(
-                argument_filters=argument_filters,
-                from_block=from_block,
-                to_block=to_block,
-                address=address,
-                topics=topics,
-            ),
-        )
-
-    @classmethod
-    def create_filter(  # type: ignore
-        cls: Type["ExampleFlipContractEvent"],
+        self,
         *,  # PEP 3102
         argument_filters: dict[str, Any] | None = None,
         from_block: BlockIdentifier | None = None,
@@ -586,16 +556,13 @@ class ExampleFlopContractEvent(ContractEvent):
     # super() get_logs and create_filter methods are generic, while our version adds values & types
     # pylint: disable=arguments-differ
 
-    # @combomethod destroys return types, so we are redefining functions as both class and instance
-    # pylint: disable=function-redefined
-
     # pylint: disable=useless-parent-delegation
     def __init__(self, *argument_names: tuple[str]) -> None:
         super().__init__(*argument_names)
 
-    # We ignore types here for function redefinition
-    def get_typed_logs(  # type: ignore
-        self: "ExampleFlopContractEvent",
+    @combomethod_typed
+    def get_logs_typed(
+        self,
         argument_filters: dict[str, Any] | None = None,
         from_block: BlockIdentifier | None = None,
         to_block: BlockIdentifier | None = None,
@@ -621,19 +588,10 @@ def get_typed_logs(  # type: ignore
             for abi_event in abi_events
         ]
 
-    @classmethod
-    # We ignore types here for function redefinition
-    def get_typed_logs(  # type: ignore
-        cls: Type["ExampleFlopContractEvent"],
-        argument_filters: dict[str, Any] | None = None,
-        from_block: BlockIdentifier | None = None,
-        to_block: BlockIdentifier | None = None,
-        block_hash: HexBytes | None = None,
-    ) -> Iterable[FlopEvent]:
-        """Extension of `get_logs` that return a typed dataclass of the event."""
-        abi_events = super().get_logs(
-            argument_filters=argument_filters, from_block=from_block, to_block=to_block, block_hash=block_hash
-        )
+    @combomethod_typed
+    def process_receipt_typed(self, txn_receipt: TxReceipt, errors: EventLogErrorFlags = WARN) -> Iterable[FlopEvent]:
+        """Extension of `process_receipt` that return a typed dataclass of the event."""
+        abi_events = super().process_receipt(txn_receipt, errors)
         # TODO there may be issues with this function if the user uses a middleware that changes event structure.
         return [
             FlopEvent(
@@ -650,29 +608,9 @@ def get_typed_logs(  # type: ignore
             for abi_event in abi_events
         ]
 
+    @combomethod_typed
     def create_filter(  # type: ignore
-        self: "ExampleFlopContractEvent",
-        *,  # PEP 3102
-        argument_filters: dict[str, Any] | None = None,
-        from_block: BlockIdentifier | None = None,
-        to_block: BlockIdentifier = "latest",
-        address: ChecksumAddress | None = None,
-        topics: Sequence[Any] | None = None,
-    ) -> LogFilter:
-        return cast(
-            LogFilter,
-            super().create_filter(
-                argument_filters=argument_filters,
-                from_block=from_block,
-                to_block=to_block,
-                address=address,
-                topics=topics,
-            ),
-        )
-
-    @classmethod
-    def create_filter(  # type: ignore
-        cls: Type["ExampleFlopContractEvent"],
+        self,
         *,  # PEP 3102
         argument_filters: dict[str, Any] | None = None,
         from_block: BlockIdentifier | None = None,
@@ -695,9 +633,9 @@ def create_filter(  # type: ignore
 class ExampleContractEvents(ContractEvents):
     """ContractEvents for the Example contract."""
 
-    Flip: ExampleFlipContractEvent
+    Flip: Type[ExampleFlipContractEvent]
 
-    Flop: ExampleFlopContractEvent
+    Flop: Type[ExampleFlopContractEvent]
 
     def __init__(
         self,
@@ -707,21 +645,18 @@ def __init__(
     ) -> None:
         super().__init__(abi, w3, address)
         self.Flip = cast(
-            ExampleFlipContractEvent,
+            Type[ExampleFlipContractEvent],
             ExampleFlipContractEvent.factory("Flip", w3=w3, contract_abi=abi, address=address, event_name="Flip"),
         )
         self.Flop = cast(
-            ExampleFlopContractEvent,
+            Type[ExampleFlopContractEvent],
             ExampleFlopContractEvent.factory("Flop", w3=w3, contract_abi=abi, address=address, event_name="Flop"),
         )
 
 
 class ExampleWrongChoiceContractError:
     """ContractError for WrongChoice."""
 
-    # @combomethod destroys return types, so we are redefining functions as both class and instance
-    # pylint: disable=function-redefined
-
     # 4 byte error selector
     selector: str
     # error signature, i.e. CustomError(uint256,bool)
@@ -734,8 +669,9 @@ def __init__(
         self.selector = "0xc13b30d4"
         self.signature = "WrongChoice(uint8,string)"
 
-    def decode_error_data(  # type: ignore
-        self: "ExampleWrongChoiceContractError",
+    @combomethod_typed
+    def decode_error_data(
+        self,
         data: HexBytes,
         # TODO: instead of returning a tuple, return a dataclass with the input names and types just like we do for functions
     ) -> tuple[Any, ...]:
@@ -749,21 +685,6 @@ def decode_error_data(  # type: ignore
         decoded = abi_codec.decode(types, data)
         return decoded
 
-    @classmethod
-    def decode_error_data(  # type: ignore
-        cls: Type["ExampleWrongChoiceContractError"],
-        data: HexBytes,
-    ) -> tuple[Any, ...]:
-        """Decodes error data returns from a smart contract."""
-        error_abi = cast(
-            ABIFunction,
-            [item for item in example_abi if item.get("name") == "WrongChoice" and item.get("type") == "error"][0],
-        )
-        types = get_abi_input_types(error_abi)
-        abi_codec = ABICodec(default_registry)
-        decoded = abi_codec.decode(types, data)
-        return decoded
-
 
 class ExampleContractErrors:
     """ContractErrors for the Example contract."""

diff --git a/example/types/ExampleTypes.py b/example/types/ExampleTypes.py
@@ -1,6 +1,6 @@
 """Dataclasses for all structs in the Example contract.
 
-DO NOT EDIT.  This file was generated by pypechain v0.0.40.
+DO NOT EDIT.  This file was generated by pypechain v0.0.41.
 See documentation at https://github.com/delvtech/pypechain """
 
 # super() call methods are generic, while our version adds values & types
@@ -20,15 +20,15 @@
 
 from dataclasses import dataclass
 
-from pypechain.core import BaseEvent, ErrorInfo, ErrorParams
+from pypechain.core import BaseEvent, BaseEventArgs, ErrorInfo, ErrorParams
 
 
-@dataclass(kw_only=True)
+@dataclass(kw_only=True, frozen=True)
 class FlipEvent(BaseEvent):
     """The event type for event Flip"""
 
-    @dataclass
-    class FlipEventArgs:
+    @dataclass(kw_only=True, frozen=True)
+    class FlipEventArgs(BaseEventArgs):
         """The args to the event Flip"""
 
         flip: int
@@ -38,12 +38,12 @@ class FlipEventArgs:
     __name__: str = "Flip"
 
 
-@dataclass(kw_only=True)
+@dataclass(kw_only=True, frozen=True)
 class FlopEvent(BaseEvent):
     """The event type for event Flop"""
 
-    @dataclass
-    class FlopEventArgs:
+    @dataclass(kw_only=True, frozen=True)
+    class FlopEventArgs(BaseEventArgs):
         """The args to the event Flop"""
 
         flop: int

diff --git a/example/types/__init__.py b/example/types/__init__.py
@@ -1,6 +1,6 @@
 """Export all types from generated files.
 
-DO NOT EDIT.  This file was generated by pypechain v0.0.40.
+DO NOT EDIT.  This file was generated by pypechain v0.0.41.
 See documentation at https://github.com/delvtech/pypechain """
 
 from .ExampleContract import ExampleContract
diff --git a/example/types/pypechain.version b/example/types/pypechain.version
@@ -1 +1 @@
-pypechain == 0.0.40
+pypechain == 0.0.41
diff --git a/pypechain/core/__init__.py b/pypechain/core/__init__.py
@@ -1,5 +1,6 @@
 """Core pypechain functions used by generated files"""
 
-from .base_event import BaseEvent
+from .base_event import BaseEvent, BaseEventArgs
+from .combomethod_typed import combomethod_typed
 from .error import ErrorInfo, ErrorParams
 from .utilities import dataclass_to_tuple, get_abi_input_types, rename_returned_types, tuple_to_dataclass
diff --git a/pypechain/core/base_event.py b/pypechain/core/base_event.py
@@ -1,12 +1,19 @@
 """Defines the base event class for all subclasses of events."""
 
+from __future__ import annotations
+
 from dataclasses import dataclass
 
 from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
 
 
-@dataclass(kw_only=True)
+@dataclass(kw_only=True, frozen=True)
+class BaseEventArgs:
+    """The base event argument class for all subclasses of event args."""
+
+
+@dataclass(kw_only=True, frozen=True)
 class BaseEvent:
     """The base event class for all subclasses of events."""
 
@@ -19,4 +26,5 @@ class BaseEvent:
     address: ChecksumAddress
     block_hash: HexBytes
     block_number: int
+    args: BaseEventArgs
     __name__: str = "BaseEvent"
diff --git a/pypechain/core/combomethod_typed.py b/pypechain/core/combomethod_typed.py
@@ -0,0 +1,39 @@
+"""A typed version of combomethod from eth_utils."""
+
+import functools
+from typing import Any, Callable, Concatenate, Generic, Optional, ParamSpec, Type, TypeVar
+
+# TODO remove this file once https://github.com/ethereum/eth-utils/pull/264 gets merged
+
+# We use generics to define the structure of the wrapped method
+# Here, `T` is the `self` or `cls` object, `P` is the parameter of the wrapped function, and
+# `R` is the return type
+
+T = TypeVar("T")
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+# We define the generic that attaches to the function we're decorating
+# so here, P and R are the types of the parameters and return of the decorated function
+class combomethod_typed(Generic[P, R]):  # pylint: disable=invalid-name
+    """A typed version of combomethod from eth_utils."""
+
+    # The callable `Any` takes place of the obj or class
+    method: Callable[Concatenate[Any, P], R]
+
+    # The method passed in has a spot for obj or class in `Any`
+    def __init__(self, method: Callable[Concatenate[Any, P], R]) -> None:
+        self.method = method
+
+    # The getter allows for logic to call either cls or obj method
+    # with the original type decorators in the output wrapper function
+    def __get__(self, obj: Optional[T] = None, objtype: Optional[Type[T]] = None) -> Callable[P, R]:
+        # The _wrapper function is unchanged from eth-utils
+        @functools.wraps(self.method)
+        def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+            if obj is not None:
+                return self.method(obj, *args, **kwargs)
+            return self.method(objtype, *args, **kwargs)
+
+        return _wrapper