Adding find method

Author: apetenchea

arangoasync/aql.py

@@ -327,7 +327,7 @@ def response_handler(resp: Response) -> Cursor:
             if not resp.is_success:
                 raise AQLQueryExecuteError(resp, request)
             if self._executor.context == "async":
-                # We cannot have a cursor getting back async jobs
+                # We cannot have a cursor giving back async jobs
                 executor: NonAsyncExecutor = DefaultApiExecutor(
                     self._executor.connection
                 )

arangoasync/collection.py

@@ -1,8 +1,9 @@
 __all__ = ["Collection", "StandardCollection"]
 
 
-from typing import Generic, List, Optional, Sequence, Tuple, TypeVar, cast
+from typing import Any, Generic, List, Optional, Sequence, Tuple, TypeVar, cast
 
+from arangoasync.cursor import Cursor
 from arangoasync.errno import (
     DOCUMENT_NOT_FOUND,
     HTTP_BAD_PARAMETER,
@@ -25,8 +26,9 @@
     IndexGetError,
     IndexListError,
     IndexLoadError,
+    SortValidationError,
 )
-from arangoasync.executor import ApiExecutor
+from arangoasync.executor import ApiExecutor, DefaultApiExecutor, NonAsyncExecutor
 from arangoasync.request import Method, Request
 from arangoasync.response import Response
 from arangoasync.result import Result
@@ -156,6 +158,90 @@ def _prep_from_doc(
         else:
             return doc_id, {"If-Match": rev}
 
+    def _build_filter_conditions(self, filters: Optional[Json]) -> str:
+        """Build filter conditions for an AQL query.
+
+        Args:
+            filters (dict | None): Document filters.
+
+        Returns:
+            str: The complete AQL filter condition.
+        """
+        if not filters:
+            return ""
+
+        conditions = []
+        for k, v in filters.items():
+            field = k if "." in k else f"`{k}`"
+            conditions.append(f"doc.{field} == {self.serializer.dumps(v)}")
+
+        return "FILTER " + " AND ".join(conditions)
+
+    @staticmethod
+    def _is_none_or_int(obj: Any) -> bool:
+        """Check if obj is `None` or a positive integer.
+
+        Args:
+            obj: Object to check.
+
+        Returns:
+            bool: `True` if object is `None` or a positive integer.
+        """
+        return obj is None or isinstance(obj, int) and obj >= 0
+
+    @staticmethod
+    def _is_none_or_dict(obj: Any) -> bool:
+        """Check if obj is `None` or a dict.
+
+        Args:
+            obj: Object to check.
+
+        Returns:
+            bool: `True` if object is `None` or a dict.
+        """
+        return obj is None or isinstance(obj, dict)
+
+    @staticmethod
+    def _validate_sort_parameters(sort: Optional[Jsons]) -> None:
+        """Validate sort parameters for an AQL query.
+
+        Args:
+            sort (list | None): Document sort parameters.
+
+        Raises:
+            SortValidationError: If sort parameters are invalid.
+        """
+        if not sort:
+            return
+
+        for param in sort:
+            if "sort_by" not in param or "sort_order" not in param:
+                raise SortValidationError(
+                    "Each sort parameter must have 'sort_by' and 'sort_order'."
+                )
+            if param["sort_order"].upper() not in ["ASC", "DESC"]:
+                raise SortValidationError("'sort_order' must be either 'ASC' or 'DESC'")
+
+    @staticmethod
+    def _build_sort_expression(sort: Optional[Jsons]) -> str:
+        """Build a sort condition for an AQL query.
+
+        Args:
+            sort (list | None): Document sort parameters.
+
+        Returns:
+            str: The complete AQL sort condition.
+        """
+        if not sort:
+            return ""
+
+        sort_chunks = []
+        for sort_param in sort:
+            chunk = f"doc.{sort_param['sort_by']} {sort_param['sort_order']}"
+            sort_chunks.append(chunk)
+
+        return "SORT " + ", ".join(sort_chunks)
+
     @property
     def name(self) -> str:
         """Return the name of the collection.
@@ -1006,3 +1092,74 @@ def response_handler(resp: Response) -> V:
             return self._doc_deserializer.loads_many(resp.raw_body)
 
         return await self._executor.execute(request, response_handler)
+
+    async def find(
+        self,
+        filters: Optional[Json] = None,
+        skip: Optional[int] = None,
+        limit: Optional[int | str] = None,
+        allow_dirty_read: Optional[bool] = False,
+        sort: Optional[Jsons] = None,
+    ) -> Result[Cursor]:
+        """Return all documents that match the given filters.
+
+        Args:
+            filters (dict | None): Query filters.
+            skip (int | None): Number of documents to skip.
+            limit (int | str | None): Maximum number of documents to return.
+            allow_dirty_read (bool): Allow reads from followers in a cluster.
+            sort (list | None): Document sort parameters.
+
+        Returns:
+            Cursor: Document cursor.
+
+        Raises:
+            DocumentGetError: If retrieval fails.
+            SortValidationError: If sort parameters are invalid.
+        """
+        if not self._is_none_or_dict(filters):
+            raise ValueError("filters parameter must be a dict")
+        self._validate_sort_parameters(sort)
+        if not self._is_none_or_int(skip):
+            raise ValueError("skip parameter must be a non-negative int")
+        if not (self._is_none_or_int(limit) or limit == "null"):
+            raise ValueError("limit parameter must be a non-negative int")
+
+        skip = skip if skip is not None else 0
+        limit = limit if limit is not None else "null"
+        query = f"""
+            FOR doc IN @@collection
+                {self._build_filter_conditions(filters)}
+                LIMIT {skip}, {limit}
+                {self._build_sort_expression(sort)}
+                RETURN doc
+        """
+        bind_vars = {"@collection": self.name}
+        data: Json = {"query": query, "bindVars": bind_vars, "count": True}
+        headers: RequestHeaders = {}
+        if allow_dirty_read is not None:
+            if allow_dirty_read is True:
+                headers["x-arango-allow-dirty-read"] = "true"
+            else:
+                headers["x-arango-allow-dirty-read"] = "false"
+
+        request = Request(
+            method=Method.POST,
+            endpoint="/_api/cursor",
+            data=self.serializer.dumps(data),
+            headers=headers,
+        )
+
+        def response_handler(resp: Response) -> Cursor:
+            if not resp.is_success:
+                raise DocumentGetError(resp, request)
+            if self._executor.context == "async":
+                # We cannot have a cursor giving back async jobs
+                executor: NonAsyncExecutor = DefaultApiExecutor(
+                    self._executor.connection
+                )
+            else:
+                executor = cast(NonAsyncExecutor, self._executor)
+            return Cursor(executor, self.deserializer.loads(resp.raw_body))
+
+        return await self._executor.execute(request, response_handler)

arangoasync/exceptions.py

@@ -327,6 +327,10 @@ class ServerVersionError(ArangoServerError):
     """Failed to retrieve server version."""
 
 
+class SortValidationError(ArangoClientError):
+    """Invalid sort parameters."""
+
+
 class TransactionAbortError(ArangoServerError):
     """Failed to abort transaction."""
 

tests/test_document.py

@@ -10,6 +10,7 @@
     DocumentReplaceError,
     DocumentRevisionError,
     DocumentUpdateError,
+    SortValidationError,
 )
 from tests.helpers import generate_col_name
 
@@ -234,3 +235,63 @@ async def test_document_get_many(doc_col, bad_col, docs):
     # Empty list
     many = await doc_col.get_many([])
     assert len(many) == 0
+
+
+@pytest.mark.asyncio
+async def test_document_find(doc_col, bad_col, docs):
+    # Check errors first
+    with pytest.raises(DocumentGetError):
+        await bad_col.find()
+    with pytest.raises(ValueError):
+        await doc_col.find(limit=-1)
+    with pytest.raises(ValueError):
+        await doc_col.find(skip="abcd")
+    with pytest.raises(ValueError):
+        await doc_col.find(filters="abcd")
+    with pytest.raises(SortValidationError):
+        await doc_col.find(sort="abcd")
+    with pytest.raises(SortValidationError):
+        await doc_col.find(sort=[{"x": "text", "sort_order": "ASC"}])
+
+    # Insert all documents
+    await asyncio.gather(*[doc_col.insert(doc) for doc in docs])
+
+    # Empty find
+    filter_docs = []
+    async for doc in await doc_col.find():
+        filter_docs.append(doc)
+    assert len(filter_docs) == len(docs)
+
+    # Test with filter
+    filter_docs = []
+    async for doc in await doc_col.find(filters={"val": 42}):
+        filter_docs.append(doc)
+    assert len(filter_docs) == 0
+    async for doc in await doc_col.find(filters={"text": "foo"}):
+        filter_docs.append(doc)
+    assert len(filter_docs) == 3
+    filter_docs = []
+    async for doc in await doc_col.find(filters={"text": "foo", "val": 1}):
+        filter_docs.append(doc)
+    assert len(filter_docs) == 1
+
+    # Test with limit
+    filter_docs = []
+    async for doc in await doc_col.find(limit=2):
+        filter_docs.append(doc)
+    assert len(filter_docs) == 2
+
+    # Test with skip
+    filter_docs = []
+    async for doc in await doc_col.find(skip=2, allow_dirty_read=True):
+        filter_docs.append(doc)
+    assert len(filter_docs) == len(docs) - 2
+
+    # Test with sort
+    filter_docs = []
+    async for doc in await doc_col.find(
+        {}, sort=[{"sort_by": "text", "sort_order": "ASC"}]
+    ):
+        filter_docs.append(doc)
+    for idx in range(len(filter_docs) - 1):
+        assert filter_docs[idx]["text"] <= filter_docs[idx + 1]["text"]