asynced Filter and EventFilterBuilder classes

Author: pacrob

web3/_utils/events.py

@@ -88,6 +88,7 @@
 if TYPE_CHECKING:
     from web3 import Web3  # noqa: F401
     from web3._utils.filters import (  # noqa: F401
+        AsyncLogFilter,
         LogFilter,
     )
 
@@ -310,7 +311,7 @@ def is_indexed(arg: Any) -> bool:
 is_not_indexed = complement(is_indexed)
 
 
-class EventFilterBuilder:
+class BaseEventFilterBuilder:
     formatter = None
     _fromBlock = None
     _toBlock = None
@@ -410,6 +411,8 @@ def filter_params(self) -> FilterParams:
         }
         return valfilter(lambda x: x is not None, params)
 
+
+class EventFilterBuilder(BaseEventFilterBuilder):
     def deploy(self, w3: "Web3") -> "LogFilter":
         if not isinstance(w3, web3.Web3):
             raise ValueError(f"Invalid web3 argument: got: {w3!r}")
@@ -427,6 +430,25 @@ def deploy(self, w3: "Web3") -> "LogFilter":
         return log_filter
 
 
+class AsyncEventFilterBuilder(BaseEventFilterBuilder):
+    async def deploy(self, w3: "Web3") -> "AsyncLogFilter":
+        if not isinstance(w3, web3.Web3):
+            raise ValueError(f"Invalid web3 argument: got: {w3!r}")
+
+        for arg in AttributeDict.values(self.args):
+            arg._immutable = True
+        self._immutable = True
+
+        log_filter = await w3.eth.filter(self.filter_params)
+        log_filter = cast("AsyncLogFilter", log_filter)
+        log_filter.filter_params = self.filter_params
+        log_filter.set_data_filters(self.data_argument_values)
+        log_filter.builder = self
+        if self.formatter is not None:
+            log_filter.log_entry_formatter = self.formatter
+        return log_filter
+
+
 def initialize_event_topics(event_abi: ABIEvent) -> Union[bytes, List[Any]]:
     if event_abi["anonymous"] is False:
         # https://github.com/python/mypy/issues/4976

web3/_utils/filters.py

@@ -41,6 +41,7 @@
 )
 
 from web3._utils.events import (
+    AsyncEventFilterBuilder,
     EventFilterBuilder,
     construct_event_data_set,
     construct_event_topic_set,
@@ -62,6 +63,7 @@
 if TYPE_CHECKING:
     from web3 import Web3  # noqa: F401
     from web3.eth import Eth  # noqa: F401
+    from web3.eth import AsyncEth  # noqa: F401
 
 
 def construct_event_filter_params(
@@ -126,14 +128,13 @@ def construct_event_filter_params(
     return data_filters_set, filter_params
 
 
-class Filter:
+class BaseFilter:
     callbacks: List[Callable[..., Any]] = None
     stopped = False
     poll_interval = None
     filter_id = None
 
-    def __init__(self, filter_id: HexStr, eth_module: "Eth") -> None:
-        self.eth_module = eth_module
+    def __init__(self, filter_id: HexStr) -> None:
         self.filter_id = filter_id
         self.callbacks = []
         super().__init__()
@@ -159,6 +160,23 @@ def _filter_valid_entries(
     ) -> Iterator[LogReceipt]:
         return filter(self.is_valid_entry, entries)
 
+    def _format_log_entries(
+        self, log_entries: Optional[Iterator[LogReceipt]] = None
+    ) -> List[LogReceipt]:
+        if log_entries is None:
+            return []
+
+        formatted_log_entries = [
+            self.format_entry(log_entry) for log_entry in log_entries
+        ]
+        return formatted_log_entries
+
+
+class Filter(BaseFilter):
+    def __init__(self, filter_id: HexStr, eth_module: "Eth") -> None:
+        self.eth_module = eth_module
+        super(Filter, self).__init__(filter_id)
+
     def get_new_entries(self) -> List[LogReceipt]:
         log_entries = self._filter_valid_entries(
             self.eth_module.get_filter_changes(self.filter_id)
@@ -171,26 +189,39 @@ def get_all_entries(self) -> List[LogReceipt]:
         )
         return self._format_log_entries(log_entries)
 
-    def _format_log_entries(
-        self, log_entries: Optional[Iterator[LogReceipt]] = None
-    ) -> List[LogReceipt]:
-        if log_entries is None:
-            return []
 
-        formatted_log_entries = [
-            self.format_entry(log_entry) for log_entry in log_entries
-        ]
-        return formatted_log_entries
+class AsyncFilter(BaseFilter):
+    def __init__(self, filter_id: HexStr, eth_module: "AsyncEth") -> None:
+        self.eth_module = eth_module
+        super(AsyncFilter, self).__init__(filter_id)
+
+    async def get_new_entries(self) -> List[LogReceipt]:
+        filter_changes = await self.eth_module.get_filter_changes(self.filter_id)
+        log_entries = self._filter_valid_entries(filter_changes)
+        return self._format_log_entries(log_entries)
+
+    async def get_all_entries(self) -> List[LogReceipt]:
+        filter_logs = await self.eth_module.get_filter_logs(self.filter_id)
+        log_entries = self._filter_valid_entries(filter_logs)
+        return self._format_log_entries(log_entries)
 
 
 class BlockFilter(Filter):
     pass
 
 
+class AsyncBlockFilter(AsyncFilter):
+    pass
+
+
 class TransactionFilter(Filter):
     pass
 
 
+class AsyncTransactionFilter(AsyncFilter):
+    pass
+
+
 class LogFilter(Filter):
     data_filter_set = None
     data_filter_set_regex = None
@@ -233,6 +264,48 @@ def is_valid_entry(self, entry: LogReceipt) -> bool:
         return bool(self.data_filter_set_function(entry["data"]))
 
 
+class AsyncLogFilter(AsyncFilter):
+    data_filter_set = None
+    data_filter_set_regex = None
+    data_filter_set_function = None
+    log_entry_formatter = None
+    filter_params: FilterParams = None
+    builder: AsyncEventFilterBuilder = None
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        self.log_entry_formatter = kwargs.pop(
+            "log_entry_formatter",
+            self.log_entry_formatter,
+        )
+        if "data_filter_set" in kwargs:
+            self.set_data_filters(kwargs.pop("data_filter_set"))
+        super().__init__(*args, **kwargs)
+
+    def format_entry(self, entry: LogReceipt) -> LogReceipt:
+        if self.log_entry_formatter:
+            return self.log_entry_formatter(entry)
+        return entry
+
+    def set_data_filters(
+        self, data_filter_set: Collection[Tuple[TypeStr, Any]]
+    ) -> None:
+        """Sets the data filters (non indexed argument filters)
+
+        Expects a set of tuples with the type and value, e.g.:
+        (('uint256', [12345, 54321]), ('string', ('a-single-string',)))
+        """
+        self.data_filter_set = data_filter_set
+        if any(data_filter_set):
+            self.data_filter_set_function = match_fn(
+                self.eth_module.codec, data_filter_set
+            )
+
+    def is_valid_entry(self, entry: LogReceipt) -> bool:
+        if not self.data_filter_set:
+            return True
+        return bool(self.data_filter_set_function(entry["data"]))
+
+
 def decode_utf8_bytes(value: bytes) -> str:
     return value.decode("utf-8")