add watching of entity creation events (http polling)

Author: matthiaszimmermann

src/arkiv/__init__.py

@@ -2,6 +2,7 @@
 
 from .account import NamedAccount
 from .client import Arkiv
+from .events import EventFilter
 from .node import ArkivNode
 from .types import (
     CreateEvent,
@@ -17,6 +18,7 @@
     "ArkivNode",
     "CreateEvent",
     "DeleteEvent",
+    "EventFilter",
     "ExtendEvent",
     "NamedAccount",
     "TransactionReceipt",

src/arkiv/client.py

@@ -128,12 +128,17 @@ def __exit__(
         exc_val: BaseException | None,
         exc_tb: Any,
     ) -> None:
+        # Cleanup event filters first
+        logger.debug("Cleaning up event filters...")
+        self.arkiv.cleanup_filters()
+
+        # Then stop the node if managed
         if self.node:
             logger.debug("Stopping managed ArkivNode...")
             self.node.stop()
 
     def __del__(self) -> None:
-        if self.node and self.node.is_running():
+        if self.node and self.node.is_running:
             logger.warning(
                 "Arkiv client with managed node is being destroyed but node is still running. "
                 "Call arkiv.node.stop() or use context manager: 'with Arkiv() as arkiv:'"

src/arkiv/events.py

@@ -0,0 +1,182 @@
+"""Event filtering for Arkiv entity events."""
+
+import logging
+import threading
+import time
+from typing import Any, cast
+
+from eth_typing import HexStr
+from web3._utils.filters import LogFilter
+from web3.contract import Contract
+from web3.contract.contract import ContractEvent
+from web3.types import EventData, LogReceipt
+
+from .contract import CREATED_EVENT
+from .types import CreateCallback, CreateEvent, EventType, TxHash
+from .utils import to_entity_key
+
+logger = logging.getLogger(__name__)
+
+
+class EventFilter:
+    """Handle for watching entity events."""
+
+    def __init__(
+        self,
+        contract: Contract,
+        event_type: EventType,
+        callback: CreateCallback,
+        from_block: str | int = "latest",
+        auto_start: bool = True,
+    ) -> None:
+        """
+        Initialize event filter.
+
+        Args:
+            contract: Web3 contract instance
+            event_type: Type of event to watch
+            callback: Callback function for the event
+            from_block: Starting block for the filter
+            auto_start: If True, starts polling immediately
+        """
+        self.contract: Contract = contract
+        self.event_type: EventType = event_type
+        self.callback: CreateCallback = callback
+        self.from_block: str | int = from_block
+
+        # Internal state
+        self._filter: LogFilter | None = None
+        self._running: bool = False
+        self._thread: threading.Thread | None = None
+        self._poll_interval: float = 2.0  # seconds
+
+        if auto_start:
+            self.start()
+
+    def start(self) -> None:
+        """
+        Start polling for events.
+        """
+        if self._running:
+            logger.warning(f"Filter for {self.event_type} is already running")
+            return
+
+        logger.info(f"Starting event filter for {self.event_type}")
+
+        # Create the Web3 filter
+        if self.event_type == "created":
+            event: ContractEvent = self.contract.events[CREATED_EVENT]
+            self._filter = event.create_filter(from_block=self.from_block)
+        else:
+            raise NotImplementedError(
+                f"Event type {self.event_type} not yet implemented"
+            )
+
+        # Start polling thread
+        self._running = True
+        self._thread = threading.Thread(target=self._poll_loop, daemon=True)
+        self._thread.start()
+
+        logger.info(f"Event filter for {self.event_type} started")
+
+    def stop(self) -> None:
+        """
+        Stop polling for events.
+        """
+        if not self._running:
+            logger.warning(f"Filter for {self.event_type} is not running")
+            return
+
+        logger.info(f"Stopping event filter for {self.event_type}")
+        self._running = False
+
+        if self._thread:
+            self._thread.join(timeout=5.0)
+            self._thread = None
+
+        logger.info(f"Event filter for {self.event_type} stopped")
+
+    @property
+    def is_running(self) -> bool:
+        """
+        Check if the filter is currently running.
+
+        Returns:
+            True if the filter's polling loop is active, False otherwise
+        """
+        return self._running
+
+    def uninstall(self) -> None:
+        """Uninstall the filter and cleanup resources."""
+        logger.info(f"Uninstalling event filter for {self.event_type}")
+
+        # Stop polling if running
+        if self._running:
+            self.stop()
+
+        # Clear filter reference (Web3 filters don't have uninstall method)
+        self._filter = None
+
+        logger.info(f"Event filter for {self.event_type} uninstalled")
+
+    def _poll_loop(self) -> None:
+        """Background polling loop for events."""
+        logger.debug(f"Poll loop started for {self.event_type}")
+
+        while self._running:
+            try:
+                # Get new entries from filter
+                if self._filter:
+                    new_entries: list[LogReceipt] = self._filter.get_new_entries()
+
+                    for entry in new_entries:
+                        try:
+                            # LogFilter from contract event has log_entry_formatter that
+                            # converts LogReceipt to EventData, but type system shows LogReceipt
+                            self._process_event(cast(EventData, entry))
+                        except Exception as e:
+                            logger.error(
+                                f"Error processing event: {e}", exc_info=True
+                            )
+
+                # Sleep before next poll
+                time.sleep(self._poll_interval)
+
+            except Exception as e:
+                logger.error(f"Error in poll loop: {e}", exc_info=True)
+                time.sleep(self._poll_interval)
+
+        logger.debug(f"Poll loop ended for {self.event_type}")
+
+    def _process_event(self, event_data: EventData) -> None:
+        """
+        Process a single event and trigger callback.
+
+        Args:
+            event_data: Event data from Web3 filter
+        """
+        logger.info(f"Processing event: {event_data}")
+
+        # Extract data based on event type
+        if self.event_type == "created":
+            # Parse CreateEvent
+            entity_key = to_entity_key(event_data["args"]["entityKey"])
+            expiration_block = event_data["args"]["expirationBlock"]
+            
+            # Ensure transaction hash has 0x prefix
+            tx_hash_hex = event_data["transactionHash"].hex()
+            if not tx_hash_hex.startswith("0x"):
+                tx_hash_hex = f"0x{tx_hash_hex}"
+            tx_hash = TxHash(HexStr(tx_hash_hex))
+
+            event = CreateEvent(
+                entity_key=entity_key, expiration_block=expiration_block
+            )
+
+            # Trigger callback
+            try:
+                self.callback(event, tx_hash)
+            except Exception as e:
+                logger.error(f"Error in callback: {e}", exc_info=True)
+        else:
+            logger.warning(f"Unknown event type: {self.event_type}")

src/arkiv/module.py

@@ -11,12 +11,14 @@
 from arkiv.account import NamedAccount
 
 from .contract import EVENTS_ABI, FUNCTIONS_ABI, STORAGE_ADDRESS
+from .events import EventFilter
 from .types import (
     ALL,
     ANNOTATIONS,
     METADATA,
     PAYLOAD,
     Annotations,
+    CreateCallback,
     CreateOp,
     DeleteOp,
     Entity,
@@ -55,6 +57,9 @@ def __init__(self, client: "Arkiv") -> None:
         for event in self.contract.all_events():
             logger.debug(f"Entity event {event.topic}: {event.signature}")
 
+        # Track active event filters for cleanup
+        self._active_filters: list[EventFilter] = []
+
     def is_available(self) -> bool:
         """Check if Arkiv functionality is available. Should always be true for Arkiv clients."""
         return True
@@ -408,3 +413,72 @@ def _get_entity_metadata(self, entity_key: EntityKey) -> dict[str, Any]:
         metadata: dict[str, Any] = self.client.eth.get_entity_metadata(entity_key)  # type: ignore[attr-defined]
         logger.debug(f"Raw metadata: {metadata}")
         return metadata
+
+    def watch_entity_created(
+        self,
+        callback: CreateCallback,
+        *,
+        from_block: str | int = "latest",
+        auto_start: bool = True,
+    ) -> EventFilter:
+        """
+        Watch for entity creation events.
+
+        Args:
+            callback: Function to call when a creation event is detected.
+                     Receives (CreateEvent, TxHash) as arguments.
+            from_block: Starting block for the filter ('latest' or block number)
+            auto_start: If True, starts polling immediately
+
+        Returns:
+            EventFilter instance for controlling the watch
+
+        Example:
+            def on_create(event: CreateEvent, tx_hash: TxHash) -> None:
+                print(f"Entity created: {event.entity_key}")
+
+            filter = arkiv.watch_entity_created(on_create)
+            # ... later ...
+            filter.stop()
+        """
+        event_filter = EventFilter(
+            contract=self.contract,
+            event_type="created",
+            callback=callback,
+            from_block=from_block,
+            auto_start=auto_start,
+        )
+
+        # Track the filter for cleanup
+        self._active_filters.append(event_filter)
+
+        return event_filter
+
+    @property
+    def active_filters(self) -> list[EventFilter]:
+        """Get a copy of currently active event filters."""
+        return list(self._active_filters)
+
+    def cleanup_filters(self) -> None:
+        """
+        Stop and uninstall all active event filters.
+
+        This is automatically called when the Arkiv client exits its context,
+        but can be called manually if needed.
+        """
+        if not self._active_filters:
+            logger.debug("No active filters to cleanup")
+            return
+
+        logger.info(
+            f"Cleaning up {len(self._active_filters)} active event filter(s)..."
+        )
+
+        for event_filter in self._active_filters:
+            try:
+                event_filter.uninstall()
+            except Exception as e:
+                logger.warning(f"Error cleaning up filter: {e}")
+
+        self._active_filters.clear()
+        logger.info("All event filters cleaned up")

src/arkiv/node.py

@@ -219,6 +219,7 @@ def container(self) -> DockerContainer:
             )
         return self._container
 
+    @property
     def is_external(self) -> bool:
         """
         Check if this node is configured as an external node.
@@ -228,6 +229,7 @@ def is_external(self) -> bool:
         """
         return self._is_external
 
+    @property
     def is_running(self) -> bool:
         """
         Check if the node is currently running.
@@ -351,7 +353,7 @@ def fund_account(self, account: NamedAccount) -> None:
                 f"Cannot fund account on external node - account {account.name} must be pre-funded via external means"
             )
 
-        if not self.is_running():
+        if not self.is_running:
             msg = "Node is not running. Call start() first."
             raise RuntimeError(msg)
 

src/arkiv/provider.py

@@ -137,7 +137,7 @@ def node(self, arkiv_node: ArkivNode | None = None) -> ProviderBuilder:
             arkiv_node = ArkivNode()
 
         # Auto-start the node if not running
-        if not arkiv_node.is_running():
+        if not arkiv_node.is_running:
             logger.debug("Auto-starting managed ArkivNode...")
             arkiv_node.start()
 

src/arkiv/types.py

@@ -1,8 +1,8 @@
-"""Arkiv SDK Types."""
+"""Type definitions for the Arkiv SDK."""
 
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from dataclasses import dataclass
-from typing import NewType
+from typing import Literal, NewType
 
 from eth_typing import ChecksumAddress, HexStr
 from web3.datastructures import AttributeDict
@@ -181,6 +181,15 @@ class TransactionReceipt:
     deletes: Sequence[DeleteEvent]
 
 
+# Event callback types
+CreateCallback = Callable[[CreateEvent, TxHash], None]
+UpdateCallback = Callable[[UpdateEvent, TxHash], None]
+DeleteCallback = Callable[[DeleteEvent, TxHash], None]
+ExtendCallback = Callable[[ExtendEvent, TxHash], None]
+
+# Event type literal
+EventType = Literal["created", "updated", "deleted", "extended"]
+
 # Low level annotations for RLP encoding
 StringAnnotationsRlp = NewType("StringAnnotationsRlp", list[tuple[str, str]])
 NumericAnnotationsRlp = NewType("NumericAnnotationsRlp", list[tuple[str, int]])

tests/conftest.py

@@ -114,7 +114,7 @@ def arkiv_node() -> Generator[ArkivNode, None, None]:
 def arkiv_client_http(arkiv_node: ArkivNode, account_1: NamedAccount) -> Arkiv:
     """Return Arkiv client with funded account connected via HTTP."""
     # Fund the account using the node (only for local containerized nodes)
-    if not arkiv_node.is_external():
+    if not arkiv_node.is_external:
         arkiv_node.fund_account(account_1)
 
     # Create provider and client
@@ -151,7 +151,7 @@ def account_2(arkiv_node: ArkivNode) -> NamedAccount:
     account = create_account(2, BOB)
 
     # Fund the account using the node (only for local containerized nodes)
-    if not arkiv_node.is_external():
+    if not arkiv_node.is_external:
         arkiv_node.fund_account(account)
 
     return account