Contract utils for get_event_by_signature, get_event_by_selector, get_event_by_topic

Author: Stuart Reed

docs/web3.contract.rst

@@ -345,6 +345,18 @@ Each Contract Factory exposes the following methods.
         [<Event Transfer(uint256,bool)>, <Event Transfer(int256,bool)>]
 
 
+.. py:classmethod:: Contract.get_event_by_signature(signature)
+
+    Searches for a distinct event with matching signature. Returns an instance of
+    :py:class:`ContractEvent` upon finding a match. Raises ``Web3ValueError`` if no
+    match is found.
+
+    .. code-block:: python
+
+        >>> contract.get_event_by_signature('Transfer(uint256,bool)')
+        <Event Transfer(uint256,bool)>
+
+
 .. py:classmethod:: Contract.find_events_by_name(name)
 
     Searches for all events matching the provided name. Returns a list of matching
@@ -368,6 +380,55 @@ Each Contract Factory exposes the following methods.
         >>> contract.get_event_by_name('UniqueEvent')
         <Event UniqueEvent(uint256)>
 
+.. py:classmethod:: Contract.find_events_by_selector(selector)
+
+    Searches for all events matching the provided selector. Returns a list of matching
+    events where every event is an instance of :py:class:`ContractEvent`. Returns an
+    empty list when no match is found.
+
+    .. code-block:: python
+
+        >>> contract.find_events_by_selector('0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15')
+        [<Event LogSingleWithIndex(uint256)>]
+
+.. py:classmethod:: Contract.get_event_by_selector(selector)
+
+    Searches for a distinct event with matching selector.
+    The selector can be a hexadecimal string, bytes or int.
+    Returns an instance of :py:class:`ContractEvent` upon finding a match.
+    Raises ``Web3ValueError`` if no match is found.
+
+    .. code-block:: python
+
+        >>> contract.get_event_by_selector('0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15')
+        <Event LogSingleWithIndex(uint256)>
+        >>> contract.get_event_by_selector(b"\xf7\x0f\xe6\x89\xe2\x90\xd8\xce+*8\x8a\xc2\x8d\xb3o\xbb\x0e\x16\xa6\xd8\x9ch\x04\xc4a\xf6Z\x1b@\xbb\x15")
+        <Event LogSingleWithIndex(uint256)>
+        >>> contract.get_event_by_selector(0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15)
+        <Event LogSingleWithIndex(uint256)>
+
+.. py:classmethod:: Contract.find_events_by_topic(topic)
+
+    Searches for all events matching the provided topic. Returns a list of matching
+    events where every event is an instance of :py:class:`ContractEvent`. Returns an
+    empty list when no match is found.
+
+    .. code-block:: python
+
+        >>> contract.find_events_by_topic('0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15')
+        [<Event LogSingleWithIndex(uint256)>]
+
+.. py:classmethod:: Contract.get_event_by_topic(topic)
+
+    Searches for a distinct event with matching topic.
+    The topic is a hexadecimal string.
+    Returns an instance of :py:class:`ContractEvent` upon finding a match.
+    Raises ``Web3ValueError`` if no match is found.
+
+    .. code-block:: python
+
+        >>> contract.get_event_by_topic('0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15')
+        <Event LogSingleWithIndex(uint256)>
 
 .. py:classmethod:: Contract.all_functions()
 

tests/core/contracts/test_contract_events.py

@@ -39,6 +39,12 @@
                 "<Event LogSingleWithIndex(uint256)>",
             ],
         ),
+        (
+            "get_event_by_signature",
+            ("LogSingleArg(uint256)",),
+            repr,
+            "<Event LogSingleArg(uint256)>",
+        ),
         (
             "find_events_by_name",
             ("LogSingleArg",),
@@ -53,6 +59,42 @@
             repr,
             "<Event LogSingleArg(uint256)>",
         ),
+        (
+            "find_events_by_selector",
+            (
+                b"\xf7\x0f\xe6\x89\xe2\x90\xd8\xce+*8\x8a\xc2\x8d\xb3o\xbb\x0e\x16\xa6\xd8\x9ch\x04\xc4a\xf6Z\x1b@\xbb\x15",  # noqa: E501
+            ),
+            map_repr,
+            [
+                "<Event LogSingleWithIndex(uint256)>",
+            ],
+        ),
+        (
+            "get_event_by_selector",
+            (
+                b"\xf7\x0f\xe6\x89\xe2\x90\xd8\xce+*8\x8a\xc2\x8d\xb3o\xbb\x0e\x16\xa6\xd8\x9ch\x04\xc4a\xf6Z\x1b@\xbb\x15",  # noqa: E501
+            ),
+            repr,
+            "<Event LogSingleWithIndex(uint256)>",
+        ),
+        (
+            "get_event_by_selector",
+            (0xF70FE689E290D8CE2B2A388AC28DB36FBB0E16A6D89C6804C461F65A1B40BB15,),
+            repr,
+            "<Event LogSingleWithIndex(uint256)>",
+        ),
+        (
+            "get_event_by_selector",
+            ("0xf70fe689e290d8ce2b2a388ac28db36fbb0e16a6d89c6804c461f65a1b40bb15",),
+            repr,
+            "<Event LogSingleWithIndex(uint256)>",
+        ),
+        (
+            "get_event_by_topic",
+            ("0x56d2ef3c5228bf5d88573621e325a4672ab50e033749a601e4f4a5e1dce905d4",),
+            repr,
+            "<Event LogSingleArg(uint256)>",
+        ),
     ),
 )
 def test_find_or_get_events_by_type(
@@ -64,8 +106,8 @@ def test_find_or_get_events_by_type(
     event_contract: "Contract",
 ) -> None:
     contract = w3.eth.contract(abi=event_contract.abi)
-    contract_function = getattr(contract, method)(*args)
-    assert repr_func(contract_function) == expected
+    contract_event = getattr(contract, method)(*args)
+    assert repr_func(contract_event) == expected
 
 
 def test_find_events_by_identifier(w3: "Web3", event_contract: "Contract") -> None:
@@ -85,7 +127,7 @@ def test_get_event_by_identifier(w3: "Web3", event_contract: "Contract") -> None
     assert repr(contract_event) == "<Event LogSingleArg(uint256)>"
 
 
-def test_functions_iterator(math_contract):
+def test_events_iterator(math_contract):
     all_events = math_contract.all_events()
     events_iter = math_contract.events
 
@@ -94,7 +136,7 @@ def test_functions_iterator(math_contract):
         assert event.name == expected_event.name
 
 
-def test_async_functions_iterator(async_math_contract):
+def test_async_events_iterator(async_math_contract):
     all_events = async_math_contract.all_events()
     events_iter = async_math_contract.events
 

web3/contract/base_contract.py

@@ -39,6 +39,8 @@
     get_normalized_abi_inputs,
     is_list_like,
     is_text,
+    keccak,
+    to_bytes,
     to_tuple,
 )
 from hexbytes import (
@@ -67,6 +69,7 @@
     empty,
 )
 from web3._utils.encoding import (
+    hexstr_if_str,
     to_4byte_hex,
     to_hex,
 )
@@ -784,7 +787,7 @@ def encode_abi(
     ) -> HexStr:
         """
         Encodes the arguments using the Ethereum ABI for the contract function
-        that matches the given name and arguments..
+        that matches the given name and arguments.
 
         :param data: defaults to function selector
         """
@@ -811,12 +814,20 @@ def encode_abi(
     def all_functions(
         self,
     ) -> "BaseContractFunction":
+        """
+        Return all functions in the contract.
+        """
         return self.find_functions_by_identifier(
             self.abi, self.w3, self.address, lambda _: True
         )
 
     @combomethod
     def get_function_by_signature(self, signature: str) -> "BaseContractFunction":
+        """
+        Return a distinct function with matching signature.
+        Raises a Web3ValueError if the signature is invalid or if there is no match or
+        more than one is found.
+        """
         if " " in signature:
             raise Web3ValueError(
                 "Function signature should not contain any spaces. "
@@ -833,6 +844,11 @@ def callable_check(fn_abi: ABIFunction) -> bool:
 
     @combomethod
     def find_functions_by_name(self, fn_name: str) -> "BaseContractFunction":
+        """
+        Return all functions with matching name.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+
         def callable_check(fn_abi: ABIFunction) -> bool:
             return fn_abi["name"] == fn_name
 
@@ -842,13 +858,22 @@ def callable_check(fn_abi: ABIFunction) -> bool:
 
     @combomethod
     def get_function_by_name(self, fn_name: str) -> "BaseContractFunction":
+        """
+        Return a distinct function with matching name.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
         fns = self.find_functions_by_name(fn_name)
         return self.get_function_by_identifier(fns, "name")
 
     @combomethod
     def get_function_by_selector(
         self, selector: Union[bytes, int, HexStr]
     ) -> "BaseContractFunction":
+        """
+        Return a distinct function with matching 4byte selector.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+
         def callable_check(fn_abi: ABIFunction) -> bool:
             return encode_hex(function_abi_to_4byte_selector(fn_abi)) == to_4byte_hex(
                 selector
@@ -863,6 +888,9 @@ def callable_check(fn_abi: ABIFunction) -> bool:
     def decode_function_input(
         self, data: HexStr
     ) -> Tuple["BaseContractFunction", Dict[str, Any]]:
+        """
+        Return a Tuple of the function selector and decoded arguments.
+        """
         func = self.get_function_by_selector(HexBytes(data)[:4])
         arguments = decode_transaction_data(
             func.abi, data, normalizers=BASE_RETURN_NORMALIZERS
@@ -871,6 +899,11 @@ def decode_function_input(
 
     @combomethod
     def find_functions_by_args(self, *args: Any) -> "BaseContractFunction":
+        """
+        Return all functions with matching args, checking each argument can be encoded
+        with the type.
+        """
+
         def callable_check(fn_abi: ABIFunction) -> bool:
             return check_if_arguments_can_be_encoded(
                 fn_abi,
@@ -885,20 +918,54 @@ def callable_check(fn_abi: ABIFunction) -> bool:
 
     @combomethod
     def get_function_by_args(self, *args: Any) -> "BaseContractFunction":
+        """
+        Return a distinct function with matching args, checking each argument can be
+        encoded with the type.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
         fns = self.find_functions_by_args(*args)
         return self.get_function_by_identifier(fns, "args")
 
     #
     #  Events API
     #
     @combomethod
-    def all_events(self) -> "BaseContractEvent":
+    def all_events(self) -> List["BaseContractEvent"]:
+        """
+        Return all events in the contract.
+        """
         return self.find_events_by_identifier(
             self.abi, self.w3, self.address, lambda _: True
         )
 
     @combomethod
-    def find_events_by_name(self, event_name: str) -> "BaseContractEvent":
+    def get_event_by_signature(self, signature: str) -> "BaseContractEvent":
+        """
+        Return a distinct event with matching signature.
+        Raises a Web3ValueError if the signature is invalid or if there is no match or
+        more than one is found.
+        """
+        if " " in signature:
+            raise Web3ValueError(
+                "Event signature should not contain any spaces. "
+                f"Found spaces in input: {signature}"
+            )
+
+        def callable_check(event_abi: ABIEvent) -> bool:
+            return abi_to_signature(event_abi) == signature
+
+        events = self.find_events_by_identifier(
+            self.abi, self.w3, self.address, callable_check
+        )
+        return self.get_event_by_identifier(events, "signature")
+
+    @combomethod
+    def find_events_by_name(self, event_name: str) -> List["BaseContractEvent"]:
+        """
+        Return all events with matching name.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+
         def callable_check(fn_abi: ABIFunction) -> bool:
             return fn_abi["name"] == event_name
 
@@ -908,9 +975,68 @@ def callable_check(fn_abi: ABIFunction) -> bool:
 
     @combomethod
     def get_event_by_name(self, event_name: str) -> "BaseContractEvent":
+        """
+        Return a distinct event with matching name.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
         events = self.find_events_by_name(event_name)
         return self.get_event_by_identifier(events, "name")
 
+    @combomethod
+    def find_events_by_selector(
+        self, selector: Union[bytes, int, HexStr]
+    ) -> List["BaseContractEvent"]:
+        """
+        Return all events with matching selector.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+
+        def callable_check(event_abi: ABIEvent) -> bool:
+            return encode_hex(
+                keccak(text=abi_to_signature(event_abi).replace(" ", ""))
+            ) == encode_hex(hexstr_if_str(to_bytes, selector))
+
+        return self.find_events_by_identifier(
+            self.abi, self.w3, self.address, callable_check
+        )
+
+    @combomethod
+    def get_event_by_selector(
+        self, selector: Union[bytes, int, HexStr]
+    ) -> "BaseContractEvent":
+        """
+        Return a distinct event with matching keccak selector.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+        events = self.find_events_by_selector(selector)
+        return self.get_event_by_identifier(events, "selector")
+
+    @combomethod
+    def find_events_by_topic(self, topic: HexStr) -> List["BaseContractEvent"]:
+        """
+        Return all events with matching topic.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+
+        def callable_check(event_abi: ABIEvent) -> bool:
+            return (
+                encode_hex(keccak(text=abi_to_signature(event_abi).replace(" ", "")))
+                == topic
+            )
+
+        return self.find_events_by_identifier(
+            self.abi, self.w3, self.address, callable_check
+        )
+
+    @combomethod
+    def get_event_by_topic(self, topic: HexStr) -> "BaseContractEvent":
+        """
+        Return a distinct event with matching topic.
+        Raises a Web3ValueError if there is no match or more than one is found.
+        """
+        events = self.find_events_by_topic(topic)
+        return self.get_event_by_identifier(events, "topic")
+
     #
     # Private Helpers
     #

web3/contract/contract.py

@@ -589,6 +589,7 @@ def find_events_by_identifier(
     def get_event_by_identifier(
         cls, events: Sequence["ContractEvent"], identifier: str
     ) -> "ContractEvent":
+        print("there", events)
         return get_event_by_identifier(events, identifier)