diff --git a/bots/example.py b/bots/example.py index c15ab0db..86869414 100644 --- a/bots/example.py +++ b/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) diff --git a/silverback/main.py b/silverback/main.py index 916b418c..94ba5ae8 100644 --- a/silverback/main.py +++ b/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, ) diff --git a/silverback/runner.py b/silverback/runner.py index 8cca47cd..a3da2e63 100644 --- a/silverback/runner.py +++ b/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(