feat(Runner): support multiple addresses for event container

Author: fubuloubu

bots/example.py

@@ -93,7 +93,8 @@ async def handle_mints(log):
 
 # You can use generic `ContractContainer.EventType`s, to get matching logs from any contract
 # NOTE: This will match based on `event_id := keccak(event.selector)`, so any matching will work
-@bot.on_(Token.Approval, spender=ROUTER)
+# NOTE: You can filter on logs from multiple addresses using `from_addresses=`
+@bot.on_(Token.Approval, from_addresses=["YFI", "WBTC", "USDT"])
 # Any handler function can be async too
 async def exec_event2(log: ContractLog):
     token = Token.at(log.contract_address)

silverback/main.py

@@ -1,4 +1,5 @@
 import atexit
+from collections.abc import Sequence
 import inspect
 from collections import defaultdict
 from datetime import datetime, timedelta
@@ -377,6 +378,7 @@ def broker_task_decorator(
         self,
         task_type: TaskType,
         container: BlockContainer | ContractEvent | ContractEventWrapper | None = None,
+        from_addresses: Sequence[AddressType | str] | None = None,
         filter_args: dict[str, Any] | None = None,
         cron_schedule: str | None = None,
         metric_name: str | None = None,
@@ -393,6 +395,9 @@ def broker_task_decorator(
         Args:
             task_type: :class:`~silverback.types.TaskType`: The type of task to create.
             container: (BlockContainer | ContractEvent): The event source to watch.
+            from_addresses: (Sequence[AddressType | str] | None):
+                The set of addresses to filter an anonymous event by.
+                Defaults to none (matches all), ignored if `container` is not anonymous event.
 
         Returns:
             Callable[[Callable], :class:`~taskiq.AsyncTaskiqDecoratedTask`]:
@@ -456,6 +461,11 @@ def add_taskiq_task(
                 ):
                     labels["address"] = contract.address
 
+                elif from_addresses is not None:
+                    labels["address"] = ",".join(
+                        self.conversion_manager.convert(a, AddressType) for a in from_addresses
+                    )
+
                 labels["event"] = container.abi.signature
 
                 topics: list[list[HexStr] | HexStr | None] = [
@@ -602,6 +612,7 @@ def do_something_on_shutdown(state):
     def on_(
         self,
         container: BlockContainer | ContractEvent,
+        from_addresses: Sequence[AddressType | str] | None = None,
         filter_args: dict[str, Any] | None = None,
         **filter_kwargs: dict[str, Any],
     ) -> Callable[[Callable], AsyncTaskiqDecoratedTask]:
@@ -610,6 +621,9 @@ def on_(
 
         Args:
             container: (BlockContainer | ContractEvent): The event source to watch.
+            from_addresses: (Sequence[AddressType | str] | None):
+                The set of addresses to filter an anonymous event by.
+                Defaults to none (matches all), ignored if `container` is not anonymous event.
             filter_args: (dict[str, Any] | None):
                 Arguments to use for event log filter. Gets combined with ``filter_kwargs``.
                 Is useful for when an event argument name is a Python keyword.
@@ -634,6 +648,7 @@ def on_(
             return self.broker_task_decorator(
                 TaskType.EVENT_LOG,
                 container=container,
+                from_addresses=from_addresses,
                 filter_args=filter_kwargs,
             )
 

silverback/runner.py

@@ -457,19 +457,28 @@ async def log_handler(ctx: LogsSubscriptionContext):
                 )
             )
 
-        contract_address = task_data.labels.get("address")
+        if contract_addresses_str := task_data.labels.get("address"):
+            contract_addresses = list(map(to_checksum_address, contract_addresses_str.split(",")))
+
+        else:
+            contract_addresses = None
+
         topics = decode_topics_from_string(task_data.labels.get("topics", "")) or None
         sub_id = await self._web3.subscription_manager.subscribe(
             LogsSubscription(
                 label=task_data.name,
-                address=to_checksum_address(contract_address) if contract_address else None,
+                address=contract_addresses,
                 topics=topics,  # type: ignore[arg-type]
                 handler=log_handler,
             )
         )
-        logger.debug(
-            f"Handling '{contract_address or ''}:{topics[0] if topics else ''}' logs via {sub_id}"
-        )
+        if contract_addresses:
+            for address in contract_addresses:
+                logger.debug(
+                    f"Handling '{address}:{topics[0] if topics else ''}' logs via {sub_id}"
+                )
+        else:
+            logger.debug(f"Handling '*:{topics[0] if topics else ''}' logs via {sub_id}")
 
     def _daemon_tasks(self) -> list[Coroutine]:
         # NOTE: Handle this as a daemon task (after startup)
@@ -503,7 +512,17 @@ async def _block_task(self, task_data: TaskData):
             self._runtime_task_group.create_task(self.run_task(task_data, block))
 
     async def _event_task(self, task_data: TaskData):
-        contract_address = task_data.labels.get("address")
+        if contract_addresses_str := task_data.labels.get("address"):
+            contract_addresses = list(map(to_checksum_address, contract_addresses_str.split(",")))
+
+            if len(contract_addresses) != 1:
+                raise ValueError("Only 1 contract address supported for Polling runner.")
+
+            contract_address = contract_addresses[0]
+
+        else:
+            contract_address = None
+
         event = EventABI.from_signature(task_data.labels["event"])
         topics = decode_topics_from_string(task_data.labels.get("topics", "")) or None
         async for log in async_wrap_iter(