feat: adds ContractLog.topics property for calculating the topics from a log

setup.py

@@ -128,7 +128,7 @@
         "web3[tester]>=6.20.1,<8",
         # ** Dependencies maintained by ApeWorX **
         "eip712>=0.2.10,<0.3",
-        "ethpm-types>=0.6.19,<0.7",
+        "ethpm-types>=0.6.23,<0.7",
         "eth_pydantic_types>=0.1.3,<0.2",
         "evmchains>=0.1.0,<0.2",
         "evm-trace>=0.2.3,<0.3",

src/ape/types/events.py

@@ -1,11 +1,8 @@
 from collections.abc import Iterable, Iterator, Sequence
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, Optional, Union, cast
+from typing import TYPE_CHECKING, Any, Optional, Union
 
-from eth_abi.abi import encode
-from eth_abi.packed import encode_packed
-from eth_pydantic_types import HexBytes
-from eth_typing import Hash32, HexStr
+from eth_pydantic_types import HexBytes, HexStr
 from eth_utils import encode_hex, is_hex, keccak, to_hex
 from ethpm_types.abi import EventABI
 from pydantic import BaseModel, field_serializer, field_validator, model_validator
@@ -14,7 +11,7 @@
 from ape.exceptions import ContractNotFoundError
 from ape.types.address import AddressType
 from ape.types.basic import HexInt
-from ape.utils.abi import LogInputABICollection
+from ape.utils.abi import LogInputABICollection, encode_topics
 from ape.utils.basemodel import BaseInterfaceModel, ExtraAttributesMixin, ExtraModelAttributes
 from ape.utils.misc import ZERO_ADDRESS, log_instead_of_fail
 
@@ -59,13 +56,11 @@ def _convert_none_to_dict(cls, value):
         return value or {}
 
     def model_dump(self, *args, **kwargs):
-        _Hash32 = Union[Hash32, HexBytes, HexStr]
-        topics = cast(Sequence[Optional[Union[_Hash32, Sequence[_Hash32]]]], self.topic_filter)
         return FilterParams(
             address=self.addresses,
             fromBlock=to_hex(self.start_block),
             toBlock=to_hex(self.stop_block or self.start_block),
-            topics=topics,
+            topics=self.topic_filter,  # type: ignore
         )
 
     @classmethod
@@ -80,46 +75,11 @@ def from_event(
         """
         Construct a log filter from an event topic query.
         """
-        from ape import convert
-        from ape.utils.abi import LogInputABICollection, is_dynamic_sized_type
-
-        event_abi: EventABI = getattr(event, "abi", event)  # type: ignore
-        search_topics = search_topics or {}
-        topic_filter: list[Optional[HexStr]] = [encode_hex(keccak(text=event_abi.selector))]
-        abi_inputs = LogInputABICollection(event_abi)
-
-        def encode_topic_value(abi_type, value):
-            if isinstance(value, (list, tuple)):
-                return [encode_topic_value(abi_type, v) for v in value]
-            elif is_dynamic_sized_type(abi_type):
-                return encode_hex(keccak(encode_packed([str(abi_type)], [value])))
-            elif abi_type == "address":
-                value = convert(value, AddressType)
-
-            return encode_hex(encode([abi_type], [value]))
-
-        for topic in abi_inputs.topic_abi_types:
-            if topic.name in search_topics:
-                encoded_value = encode_topic_value(topic.type, search_topics[topic.name])
-                topic_filter.append(encoded_value)
-            else:
-                topic_filter.append(None)
-
-        topic_names = [i.name for i in abi_inputs.topic_abi_types if i.name]
-        invalid_topics = set(search_topics) - set(topic_names)
-        if invalid_topics:
-            raise ValueError(
-                f"{event_abi.name} defines {', '.join(topic_names)} as indexed topics, "
-                f"but you provided {', '.join(invalid_topics)}"
-            )
-
-        # remove trailing wildcards since they have no effect
-        while topic_filter[-1] is None:
-            topic_filter.pop()
-
+        abi = getattr(event, "abi", event)
+        topic_filter = encode_topics(abi, search_topics or {})
         return cls(
             addresses=addresses or [],
-            events=[event_abi],
+            events=[abi],
             topic_filter=topic_filter,
             start_block=start_block,
             stop_block=stop_block,
@@ -266,6 +226,13 @@ def abi(self) -> EventABI:
         self._abi = abi
         return abi
 
+    @cached_property
+    def topics(self) -> list[HexStr]:
+        """
+        The encoded hex-str topics values.
+        """
+        return encode_topics(self.abi, self.event_arguments)
+
     @property
     def timestamp(self) -> int:
         """

src/ape/utils/abi.py

@@ -9,12 +9,13 @@
 from eth_abi.encoding import UnsignedIntegerEncoder
 from eth_abi.exceptions import DecodingError, InsufficientDataBytes
 from eth_abi.registry import BaseEquals, registry
-from eth_pydantic_types import HexBytes
+from eth_pydantic_types import HexBytes, HexStr
 from eth_pydantic_types.validators import validate_bytes_size
 from eth_utils import decode_hex
 from ethpm_types.abi import ABIType, ConstructorABI, EventABI, EventABIType, MethodABI
 
 from ape.logging import logger
+from ape.utils.basemodel import ManagerAccessMixin
 
 ARRAY_PATTERN = re.compile(r"[(*\w,? )]*\[\d*]")
 NATSPEC_KEY_PATTERN = re.compile(r"(@\w+)")
@@ -543,3 +544,58 @@ def _enrich_natspec(natspec: str) -> str:
     # Ensure the natspec @-words are highlighted.
     replacement = r"[bright_red]\1[/]"
     return re.sub(NATSPEC_KEY_PATTERN, replacement, natspec)
+
+
+def encode_topics(abi: EventABI, topics: Optional[dict[str, Any]] = None) -> list[HexStr]:
+    """
+    Encode the given topics using the given ABI. Useful for searching logs.
+
+    Args:
+        abi (EventABI): The event.
+        topics (dict[str, Any] } None): Topic inputs to encode.
+
+    Returns:
+        list[str]: Encoded topics.
+    """
+    topics = topics or {}
+    values = {}
+
+    unnamed_iter = 0
+    topic_inputs = {}
+    for abi_input in abi.inputs:
+        if not abi_input.indexed:
+            continue
+
+        if name := abi_input.name:
+            topic_inputs[name] = abi_input
+        else:
+            topic_inputs[f"_{unnamed_iter}"] = abi_input
+
+    for input_name, input_value in topics.items():
+        if input_name not in topic_inputs:
+            # Was trying to use data or is not part of search.
+            continue
+
+        input_type = topic_inputs[input_name].type
+        if input_type == "address":
+            convert = ManagerAccessMixin.conversion_manager.convert
+            if isinstance(input_value, (list, tuple)):
+                adjusted_value = []
+                for addr in input_value:
+                    if isinstance(addr, str):
+                        adjusted_value.append(convert(addr))
+                    else:
+                        from ape.types import AddressType
+
+                        adjusted_value.append(convert(addr, AddressType))
+
+                input_value = adjusted_value
+
+            elif not isinstance(input_value, str):
+                from ape.types import AddressType
+
+                input_value = convert(input_value, AddressType)
+
+        values[input_name] = input_value
+
+    return abi.encode_topics(values)  # type: ignore

tests/functional/test_provider.py

@@ -248,6 +248,14 @@ def test_get_contract_logs_single_log(chain, contract_instance, owner, eth_teste
     logs[0]._abi = None
     assert logs[0].abi == contract_instance.FooHappened.abi
 
+    # Ensure topics are expected.
+    topics = logs[0].topics
+    expected_topics = [
+        "0x1a7c56fae0af54ebae73bc4699b9de9835e7bb86b050dff7e80695b633f17abd",
+        "0x0000000000000000000000000000000000000000000000000000000000000000",
+    ]
+    assert topics == expected_topics
+
 
 def test_get_contract_logs_single_log_query_multiple_values(
     chain, contract_instance, owner, eth_tester_provider
@@ -266,6 +274,27 @@ def test_get_contract_logs_single_log_query_multiple_values(
     assert logs[-1]["foo"] == 0
 
 
+def test_get_contract_logs_multiple_accounts_for_address(
+    chain, contract_instance, owner, eth_tester_provider
+):
+    """
+    Tests the condition when you pass in multiple AddressAPI objects
+    during an address-topic search.
+    """
+    contract_instance.logAddressArray(sender=owner)  # Create logs
+    block = chain.blocks.height
+    log_filter = LogFilter.from_event(
+        event=contract_instance.EventWithAddressArray,
+        search_topics={"some_address": [owner, contract_instance]},
+        addresses=[contract_instance, owner],
+        start_block=block,
+        stop_block=block,
+    )
+    logs = [log for log in eth_tester_provider.get_contract_logs(log_filter)]
+    assert len(logs) >= 1
+    assert logs[-1]["some_address"] == owner.address
+
+
 def test_get_contract_logs_single_log_unmatched(
     chain, contract_instance, owner, eth_tester_provider
 ):

tests/functional/test_types.py

@@ -105,6 +105,12 @@ def test_contract_log_abi(log):
     assert log.abi.name == "MyEvent"
 
 
+def test_contract_log_topics(log):
+    actual = log.topics
+    expected = ["0x4dbfb68b43dddfa12b51ebe99ab8fded620f9a0ac23142879a4f192a1b7952d2"]
+    assert actual == expected
+
+
 def test_topic_filter_encoding():
     event_abi = EventABI.model_validate_json(RAW_EVENT_ABI)
     log_filter = LogFilter.from_event(