adjust to new arkiv node for AsyncArkiv

Author: Matthias Zimmermann

src/arkiv/module.py

@@ -102,7 +102,6 @@ def create_entity(
         create_op = CreateOp(
             payload=payload, content_type=content_type, annotations=annotations, btl=btl
         )
-        logger.info(f"Creating entity:{create_op}")
 
         # Wrap in Operations container and execute
         operations = Operations(creates=[create_op])

src/arkiv/module_async.py

@@ -6,35 +6,38 @@
 import logging
 from typing import TYPE_CHECKING, Any
 
-from eth_typing import ChecksumAddress, HexStr
+from eth_typing import HexStr
 from web3.types import TxParams, TxReceipt
 
 from .events_async import AsyncEventFilter
 from .module_base import ArkivModuleBase
 from .types import (
     ALL,
-    ANNOTATIONS,
-    METADATA,
-    PAYLOAD,
+    NONE,
+    QUERY_OPTIONS_DEFAULT,
     Annotations,
     AsyncCreateCallback,
     AsyncDeleteCallback,
     AsyncExtendCallback,
     AsyncUpdateCallback,
     CreateOp,
-    Cursor,
     DeleteOp,
     Entity,
     EntityKey,
     EventType,
     ExtendOp,
     Operations,
+    QueryOptions,
     QueryResult,
     TransactionReceipt,
     TxHash,
     UpdateOp,
 )
-from .utils import merge_annotations, to_tx_params
+from .utils import (
+    to_query_result,
+    to_rpc_query_options,
+    to_tx_params,
+)
 
 # Deal with potential circular imports between client.py and module_async.py
 if TYPE_CHECKING:
@@ -75,6 +78,7 @@ async def execute(
     async def create_entity(
         self,
         payload: bytes | None = None,
+        content_type: str | None = None,
         annotations: Annotations | None = None,
         btl: int | None = None,
         tx_params: TxParams | None = None,
@@ -84,6 +88,7 @@ async def create_entity(
 
         Args:
             payload: Optional data payload for the entity
+            content_type: Optional content type of the payload
             annotations: Optional key-value annotations
             btl: Blocks to live (default: self.btl_default, ~30 minutes with 2s blocks)
             tx_params: Optional additional transaction parameters
@@ -92,10 +97,12 @@ async def create_entity(
             The entity key and transaction hash of the create operation
         """
         # Create the operation
-        payload, annotations, btl = self._check_and_set_argument_defaults(
-            payload, annotations, btl
+        payload, content_type, annotations, btl = self._check_and_set_argument_defaults(
+            payload, content_type, annotations, btl
+        )
+        create_op = CreateOp(
+            payload=payload, content_type=content_type, annotations=annotations, btl=btl
         )
-        create_op = CreateOp(payload=payload, annotations=annotations, btl=btl)
 
         # Wrap in Operations container and execute
         operations = Operations(creates=[create_op])
@@ -116,6 +123,7 @@ async def update_entity(
         self,
         entity_key: EntityKey,
         payload: bytes | None = None,
+        content_type: str | None = None,
         annotations: Annotations | None = None,
         btl: int | None = None,
         tx_params: TxParams | None = None,
@@ -134,12 +142,13 @@ async def update_entity(
             Transaction hash of the update operation
         """
         # Create the update operation
-        payload, annotations, btl = self._check_and_set_argument_defaults(
-            payload, annotations, btl
+        payload, content_type, annotations, btl = self._check_and_set_argument_defaults(
+            payload, content_type, annotations, btl
         )
         update_op = UpdateOp(
             entity_key=entity_key,
             payload=payload,
+            content_type=content_type,
             annotations=annotations,
             btl=btl,
         )
@@ -218,138 +227,81 @@ async def delete_entity(
 
         return receipt.tx_hash
 
-    async def entity_exists(self, entity_key: EntityKey) -> bool:
+    async def entity_exists(
+        self, entity_key: EntityKey, at_block: int | None = None
+    ) -> bool:
         """
-        Check if an entity exists storage (async).
+        Check if an entity exists storage.
 
         Args:
             entity_key: The entity key to check
+            at_block: Block number to pin query to specific block, or None to use latest block available
 
         Returns:
             True if the entity exists, False otherwise
         """
         try:
-            # TODO self.client.eth.get_entity_metadata by itself does not guarantee existence
-            await self._get_entity_metadata(entity_key)
-            return True
-
+            options = QueryOptions(fields=NONE, at_block=at_block)
+            query_result: QueryResult = await self.query_entities(
+                f"$key = {entity_key}", options=options
+            )
+            return len(query_result.entities) > 0
         except Exception:
             return False
 
-    async def get_entity(self, entity_key: EntityKey, fields: int = ALL) -> Entity:
+    async def get_entity(
+        self, entity_key: EntityKey, fields: int = ALL, at_block: int | None = None
+    ) -> Entity:
         """
-        Get an entity by its entity key (async).
+        Get an entity by its entity key.
 
         Args:
             entity_key: The entity key to retrieve
-            fields: Bitfield indicating which fields to retrieve
-                   PAYLOAD (1) = retrieve payload
-                   METADATA (2) = retrieve metadata
-                   ANNOTATIONS (4) = retrieve annotations
+            fields: Bitfield indicating which fields to retrieve. See file types.py
+            at_block: Block number to pin query to specific block, or None to use latest block available
 
         Returns:
             Entity object with the requested fields
         """
-        # Gather the requested data
-        owner: ChecksumAddress | None = None
-        expires_at_block: int | None = None
-        payload: bytes | None = None
-        annotations: Annotations | None = None
-
-        # HINT: rpc methods to fetch entity content might change this is the place to adapt
-        # get and decode payload if requested
-        try:
-            if fields & PAYLOAD:
-                payload = await self._get_storage_value(entity_key)
-
-            # get and decode annotations and/or metadata if requested
-            if fields & METADATA or fields & ANNOTATIONS:
-                metadata_all = await self._get_entity_metadata(entity_key)
-
-                if fields & METADATA:
-                    # Convert owner address to checksummed format
-                    owner = self._get_owner(metadata_all)
-                    expires_at_block = self._get_expires_at_block(metadata_all)
-
-                if fields & ANNOTATIONS:
-                    annotations = merge_annotations(
-                        string_annotations=metadata_all.get("stringAnnotations", []),
-                        numeric_annotations=metadata_all.get("numericAnnotations", []),
-                    )
-        except Exception as e:
-            logger.warning(f"Error fetching entity[{entity_key}]: {e}")
-
-        # Create and return entity
-        return Entity(
-            entity_key=entity_key,
-            fields=fields,
-            owner=owner,
-            expires_at_block=expires_at_block,
-            payload=payload,
-            annotations=annotations,
+
+        options = QueryOptions(fields=fields, at_block=at_block)
+        query_result: QueryResult = await self.query_entities(
+            f"$key = {entity_key}", options=options
         )
 
+        if not query_result:
+            raise ValueError(f"Entity not found: {entity_key}")
+
+        if len(query_result.entities) != 1:
+            raise ValueError(f"Expected 1 entity, got {len(query_result.entities)}")
+
+        result_entity = query_result.entities[0]
+        return result_entity
+
     async def query_entities(
         self,
         query: str | None = None,
-        *,
-        limit: int | None = None,
-        at_block: int | str = "latest",
-        cursor: Cursor | None = None,
+        options: QueryOptions = QUERY_OPTIONS_DEFAULT,
     ) -> QueryResult:
         """
-        Query entities with manual pagination control (async).
-
-        Execute a query against entity storage and return a single page of results.
-        Use the returned cursor to fetch subsequent pages.
+        Execute a query against entity storage.
 
         Args:
-            query: SQL-like query string for the first page. Mutually exclusive with cursor.
-                Example: "SELECT * WHERE owner = '0x...' ORDER BY created_at DESC"
-            limit: Maximum number of entities to return per page.
-                Server may enforce a maximum limit. Can be modified between pagination requests.
-            at_block: Block number or "latest" at which to execute query.
-                Ensures consistency across paginated results. Ignored when using cursor
-                (cursor already contains the block number).
-            cursor: Cursor from previous QueryResult to fetch next page.
-                Mutually exclusive with query. The cursor encodes the query and block number.
-
-        Returns:
-            QueryResult with entities, block number, and optional next cursor.
+            query: SQL-like query string
+            options: QueryOptions for the query execution
 
         Raises:
-            ValueError: If both query and cursor are provided, or if neither is provided.
-            NotImplementedError: This method is not yet implemented.
-
-        Examples:
-            First page:
-                >>> result = await arkiv.arkiv.query_entities(
-                ...     "SELECT * WHERE owner = '0x1234...'",
-                ...     limit=100
-                ... )
-                >>> for entity in result:
-                ...     print(entity.entity_key)
+            ValueError: if invalid query options are provided.
 
-            Next page:
-                >>> if result.has_more():
-                ...     next_page = await arkiv.arkiv.query_entities(cursor=result.next_cursor)
-
-        Note:
-            For automatic pagination across all pages, use query_all_entities() instead.
+        Returns:
+            QueryResult with entities, block number, and an optional cursor to fetch the next page
         """
         # Validate parameters using base class helper
-        self._validate_query_entities_params(query, limit, at_block, cursor)
-
-        if query:
-            # Fetch raw results from RPC
-            raw_results = await self.client.eth.query_entities(query)
-            block_number = await self.client.eth.get_block_number()
-
-            # Transform raw results into QueryResult using base class helper
-            return self._build_query_result(raw_results, block_number)
+        options.validate(query)
+        rpc_options = to_rpc_query_options(options)
+        raw_results = await self.client.eth.query(query, rpc_options)
 
-        # Cursor based pagination not implemented yet
-        raise NotImplementedError("query_entities is not yet implemented for cursors")
+        return to_query_result(options.fields, raw_results)
 
     async def watch_entity_created(
         self,