Adding support for /checksum

Author: apetenchea

arangoasync/collection.py

@@ -16,6 +16,7 @@
     HTTP_PRECONDITION_FAILED,
 )
 from arangoasync.exceptions import (
+    CollectionChecksumError,
     CollectionPropertiesError,
     CollectionResponsibleShardError,
     CollectionRevisionError,
@@ -688,6 +689,43 @@ def response_handler(resp: Response) -> str:
 
         return await self._executor.execute(request, response_handler)
 
+    async def checksum(
+        self, with_rev: Optional[bool] = None, with_data: Optional[bool] = None
+    ) -> Result[str]:
+        """Calculate collection checksum.
+
+        Args:
+            with_rev (bool | None): Include document revisions in checksum calculation.
+            with_data (bool | None): Include document data in checksum calculation.
+
+        Returns:
+            str: Collection checksum.
+
+        Raises:
+            CollectionChecksumError: If retrieval fails.
+
+        References:
+            - `get-the-collection-checksum <https://docs.arangodb.com/stable/develop/http-api/collections/#get-the-collection-checksum>`__
+        """  # noqa: E501
+        params: Params = {}
+        if with_rev is not None:
+            params["withRevision"] = with_rev
+        if with_data is not None:
+            params["withData"] = with_data
+
+        request = Request(
+            method=Method.GET,
+            endpoint=f"/_api/collection/{self.name}/checksum",
+            params=params,
+        )
+
+        def response_handler(resp: Response) -> str:
+            if not resp.is_success:
+                raise CollectionChecksumError(resp, request)
+            return cast(str, self.deserializer.loads(resp.raw_body)["checksum"])
+
+        return await self._executor.execute(request, response_handler)
+
     async def has(
         self,
         document: str | Json,

arangoasync/exceptions.py

@@ -183,6 +183,10 @@ class CollectionCreateError(ArangoServerError):
     """Failed to create collection."""
 
 
+class CollectionChecksumError(ArangoServerError):
+    """Failed to retrieve collection checksum."""
+
+
 class CollectionDeleteError(ArangoServerError):
     """Failed to delete collection."""
 

tests/test_collection.py

@@ -4,6 +4,7 @@
 
 from arangoasync.errno import DATA_SOURCE_NOT_FOUND, INDEX_NOT_FOUND
 from arangoasync.exceptions import (
+    CollectionChecksumError,
     CollectionPropertiesError,
     CollectionResponsibleShardError,
     CollectionRevisionError,
@@ -27,6 +28,8 @@ def test_collection_attributes(db, doc_col):
 
 @pytest.mark.asyncio
 async def test_collection_misc_methods(doc_col, bad_col, docs, cluster):
+    doc = await doc_col.insert(docs[0])
+
     # Properties
     properties = await doc_col.properties()
     assert properties.name == doc_col.name
@@ -44,7 +47,6 @@ async def test_collection_misc_methods(doc_col, bad_col, docs, cluster):
 
     # Shards
     if cluster:
-        doc = await doc_col.insert(docs[0])
         shard = await doc_col.responsible_shard(doc)
         assert isinstance(shard, str)
         with pytest.raises(CollectionResponsibleShardError):
@@ -60,6 +62,12 @@ async def test_collection_misc_methods(doc_col, bad_col, docs, cluster):
     with pytest.raises(CollectionRevisionError):
         await bad_col.revision()
 
+    # Checksum
+    checksum = await doc_col.checksum(with_rev=True, with_data=True)
+    assert isinstance(checksum, str)
+    with pytest.raises(CollectionChecksumError):
+        await bad_col.checksum()
+
 
 @pytest.mark.asyncio
 async def test_collection_index(doc_col, bad_col, cluster):