feat: add event filter creation and log fetching to ContractEvent

README.md

@@ -20,7 +20,6 @@ Please open issue if you want to see more ABIs included.
 
 ### TODO
 
-* event filter arguments building.
 * more tests.
 * more builtin contract ABIs for convenience
   * permit2

eth_contract/contract.py

@@ -25,10 +25,12 @@
 from typing_extensions import Unpack
 from web3 import AsyncWeb3
 from web3._utils.events import get_event_data
+from web3._utils.filters import construct_event_filter_params
 from web3.exceptions import MismatchedABI
 from web3.types import (
     BlockIdentifier,
     EventData,
+    FilterParams,
     LogReceipt,
     StateOverride,
     TxParams,
@@ -193,6 +195,79 @@ def topic(self) -> HexBytes:
             self._topic = keccak(text=self.signature)
         return self._topic
 
+    def build_filter(
+        self,
+        address: ChecksumAddress | list[ChecksumAddress] | None = None,
+        argument_filters: dict[str, Any] | None = None,
+        from_block: BlockIdentifier | None = None,
+        to_block: BlockIdentifier | None = None,
+    ) -> FilterParams:
+        """
+        Build filter parameters suitable for ``eth_getLogs``.
+
+        Args:
+            address: Contract address or list of addresses to filter by.
+            argument_filters: Mapping of indexed argument names to values to
+                filter on (e.g. ``{"from": "0x..."}``)
+            from_block: Starting block (inclusive). Defaults to the node's
+                default when omitted.
+            to_block: Ending block (inclusive). Defaults to the node's default
+                when omitted.
+
+        Returns:
+            A :class:`~web3.types.FilterParams` dict ready to be passed to
+            ``w3.eth.get_logs()``.
+        """
+        codec = ABICodec(default_registry)
+        _data_filters, filter_params = construct_event_filter_params(
+            self.abi,
+            codec,
+            contract_address=address,
+            argument_filters=argument_filters,
+            from_block=from_block,
+            to_block=to_block,
+        )
+        return filter_params
+
+    async def get_logs(
+        self,
+        w3: AsyncWeb3,
+        address: ChecksumAddress | list[ChecksumAddress] | None = None,
+        argument_filters: dict[str, Any] | None = None,
+        from_block: BlockIdentifier | None = None,
+        to_block: BlockIdentifier | None = None,
+    ) -> list[EventData]:
+        """
+        Fetch and decode matching logs from the chain.
+
+        Calls ``eth_getLogs`` using the filter built by :meth:`build_filter`
+        and decodes each returned log with :meth:`parse_log`.
+
+        Args:
+            w3: An async Web3 instance.
+            address: Contract address or list of addresses to filter by.
+            argument_filters: Mapping of indexed argument names to filter
+                values (e.g. ``{"from": "0x..."}``)
+            from_block: Starting block (inclusive).
+            to_block: Ending block (inclusive).
+
+        Returns:
+            List of decoded :class:`~web3.types.EventData` entries.
+        """
+        filter_params = self.build_filter(
+            address=address,
+            argument_filters=argument_filters,
+            from_block=from_block,
+            to_block=to_block,
+        )
+        logs = await w3.eth.get_logs(filter_params)
+        results: list[EventData] = []
+        for log in logs:
+            decoded = self.parse_log(log)
+            if decoded is not None:
+                results.append(decoded)
+        return results
+
     def parse_log(self, log: LogReceipt) -> EventData | None:
         try:
             return get_event_data(ABICodec(default_registry), self.abi, log)

tests/test_abi.py

@@ -26,3 +26,51 @@ def test_event_abi() -> None:
     assert evt.topic == HexBytes(
         "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
     )
+
+
+def test_build_filter_no_args() -> None:
+    evt = ContractEvent.from_abi(
+        "event Transfer(address indexed from, address indexed to, uint256 value)"
+    )
+    params = evt.build_filter()
+    # topics[0] must always be the event signature hash
+    assert HexBytes(params["topics"][0]) == evt.topic
+    assert "address" not in params
+    assert "fromBlock" not in params
+    assert "toBlock" not in params
+
+
+def test_build_filter_with_address() -> None:
+    evt = ContractEvent.from_abi(
+        "event Transfer(address indexed from, address indexed to, uint256 value)"
+    )
+    addr = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
+    params = evt.build_filter(address=addr)
+    assert params["address"] == addr
+    assert HexBytes(params["topics"][0]) == evt.topic
+
+
+def test_build_filter_with_block_range() -> None:
+    evt = ContractEvent.from_abi(
+        "event Transfer(address indexed from, address indexed to, uint256 value)"
+    )
+    params = evt.build_filter(from_block=100, to_block="latest")
+    assert params["fromBlock"] == 100
+    assert params["toBlock"] == "latest"
+    assert HexBytes(params["topics"][0]) == evt.topic
+
+
+def test_build_filter_with_indexed_argument() -> None:
+    evt = ContractEvent.from_abi(
+        "event Transfer(address indexed from, address indexed to, uint256 value)"
+    )
+    sender = "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
+    params = evt.build_filter(argument_filters={"from": sender})
+    topics = params["topics"]
+    # topic[0] = event signature, topic[1] = encoded `from` arg, topic[2] = None (any `to`)
+    assert HexBytes(topics[0]) == evt.topic
+    # The encoded address should be left-padded to 32 bytes
+    assert topics[1] is not None
+    assert sender[2:].lower() in topics[1].lower()  # strip 0x before substring check
+    # trailing None topics for unfiltered indexed args are stripped by web3
+    assert len(topics) == 2

tests/test_erc20.py

@@ -302,3 +302,41 @@ def test_pyrevm_trace_log():
             topics = stack[-(2 + num_topics) : -2][::-1]
             assert topics[0] == deposit_hash, "deposit event hash mismatch"
             assert to_checksum_address(topics[1]) == whale, "whale address mismatch"
+
+
+@pytest.mark.asyncio
+async def test_event_get_logs(w3):
+    """Test get_logs fetches and decodes Transfer events emitted by a mock ERC20."""
+    owner = (await w3.eth.accounts)[0]
+    salt = 300
+    initcode = get_initcode(MockERC20_ARTIFACT, "TEST", "TEST", 18)
+    token = await ensure_deployed_by_create2(w3, owner, initcode, salt=salt)
+
+    recipient = (await w3.eth.accounts)[1]
+    amt = 500
+
+    # Record block before the mint so we can filter from that point
+    from_block = await w3.eth.block_number
+
+    await ERC20.fns.mint(owner, amt).transact(w3, owner, to=token)
+    await ERC20.fns.transfer(recipient, amt).transact(w3, owner, to=token)
+
+    transfer_event = ERC20.events.Transfer
+
+    # Fetch all Transfer events for this token since the mint
+    logs = await transfer_event.get_logs(
+        w3, address=token, from_block=from_block
+    )
+    assert len(logs) >= 2
+
+    # Fetch only transfers where `from` == owner (the mint emits from == zero address,
+    # the manual transfer emits from == owner)
+    owner_logs = await transfer_event.get_logs(
+        w3,
+        address=token,
+        argument_filters={"from": owner},
+        from_block=from_block,
+    )
+    assert len(owner_logs) >= 1
+    for log in owner_logs:
+        assert log["args"]["from"] == owner