Adding /figures support

Author: apetenchea

arangoasync/collection.py

@@ -17,6 +17,7 @@
 )
 from arangoasync.exceptions import (
     CollectionPropertiesError,
+    CollectionStatisticsError,
     CollectionTruncateError,
     DocumentCountError,
     DocumentDeleteError,
@@ -41,6 +42,7 @@
 from arangoasync.serialization import Deserializer, Serializer
 from arangoasync.typings import (
     CollectionProperties,
+    CollectionStatistics,
     IndexProperties,
     Json,
     Jsons,
@@ -552,7 +554,10 @@ async def count(self) -> Result[int]:
 
         Raises:
             DocumentCountError: If retrieval fails.
-        """
+
+        References:
+            - `get-the-document-count-of-a-collection <https://docs.arangodb.com/stable/develop/http-api/collections/#get-the-document-count-of-a-collection>`__
+        """  # noqa: E501
         request = Request(
             method=Method.GET, endpoint=f"/_api/collection/{self.name}/count"
         )
@@ -565,6 +570,30 @@ def response_handler(resp: Response) -> int:
 
         return await self._executor.execute(request, response_handler)
 
+    async def statistics(self) -> Result[CollectionStatistics]:
+        """Get additional statistical information about the collection.
+
+        Returns:
+            CollectionStatistics: Collection statistics.
+
+        Raises:
+            CollectionStatisticsError: If retrieval fails.
+
+        References:
+            - `get-the-collection-statistics <https://docs.arangodb.com/stable/develop/http-api/collections/#get-the-collection-statistics>`__
+        """  # noqa: E501
+        request = Request(
+            method=Method.GET,
+            endpoint=f"/_api/collection/{self.name}/figures",
+        )
+
+        def response_handler(resp: Response) -> CollectionStatistics:
+            if not resp.is_success:
+                raise CollectionStatisticsError(resp, request)
+            return CollectionStatistics(self.deserializer.loads(resp.raw_body))
+
+        return await self._executor.execute(request, response_handler)
+
     async def has(
         self,
         document: str | Json,

arangoasync/exceptions.py

@@ -195,6 +195,14 @@ class CollectionPropertiesError(ArangoServerError):
     """Failed to retrieve collection properties."""
 
 
+class CollectionStatisticsError(ArangoServerError):
+    """Failed to retrieve collection statistics."""
+
+
+class CollectionTruncateError(ArangoServerError):
+    """Failed to truncate collection."""
+
+
 class ClientConnectionAbortedError(ArangoClientError):
     """The connection was aborted."""
 
@@ -203,10 +211,6 @@ class ClientConnectionError(ArangoClientError):
     """The request was unable to reach the server."""
 
 
-class CollectionTruncateError(ArangoServerError):
-    """Failed to truncate collection."""
-
-
 class CursorCloseError(ArangoServerError):
     """Failed to delete the cursor result from server."""
 

arangoasync/typings.py

@@ -791,8 +791,6 @@ def compatibility_formatter(data: Json) -> Json:
             result["deleted"] = data["deleted"]
         if "syncByRevision" in data:
             result["sync_by_revision"] = data["syncByRevision"]
-        if "tempObjectId" in data:
-            result["temp_object_id"] = data["tempObjectId"]
         if "usesRevisionsAsDocumentIds" in data:
             result["rev_as_id"] = data["usesRevisionsAsDocumentIds"]
         if "isDisjoint" in data:
@@ -819,6 +817,146 @@ def format(self, formatter: Optional[Formatter] = None) -> Json:
         return self.compatibility_formatter(self._data)
 
 
+class CollectionStatistics(JsonWrapper):
+    """Statistical information about the collection.
+
+    Example:
+        .. code-block:: json
+
+            {
+              "figures" : {
+                "indexes" : {
+                  "count" : 1,
+                  "size" : 1234
+                },
+                "documentsSize" : 5601,
+                "cacheInUse" : false,
+                "cacheSize" : 0,
+                "cacheUsage" : 0,
+                "engine" : {
+                  "documents" : 1,
+                  "indexes" : [
+                    {
+                      "type" : "primary",
+                      "id" : 0,
+                      "count" : 1
+                    }
+                  ]
+                }
+              },
+              "writeConcern" : 1,
+              "waitForSync" : false,
+              "usesRevisionsAsDocumentIds" : true,
+              "syncByRevision" : true,
+              "statusString" : "loaded",
+              "id" : "69123",
+              "isSmartChild" : false,
+              "schema" : null,
+              "name" : "products",
+              "type" : 2,
+              "status" : 3,
+              "count" : 1,
+              "cacheEnabled" : false,
+              "isSystem" : false,
+              "internalValidatorType" : 0,
+              "globallyUniqueId" : "hB7C02EE43DCE/69123",
+              "keyOptions" : {
+                "allowUserKeys" : true,
+                "type" : "traditional",
+                "lastValue" : 69129
+              },
+              "computedValues" : null,
+              "objectId" : "69124"
+           }
+
+    References:
+        - `get-the-collection-statistics <https://docs.arangodb.com/stable/develop/http-api/collections/#get-the-collection-statistics>`__
+    """  # noqa: E501
+
+    def __init__(self, data: Json) -> None:
+        super().__init__(data)
+
+    @property
+    def figures(self) -> Json:
+        return cast(Json, self._data.get("figures"))
+
+    @property
+    def write_concern(self) -> Optional[int]:
+        return self._data.get("writeConcern")
+
+    @property
+    def wait_for_sync(self) -> Optional[bool]:
+        return self._data.get("waitForSync")
+
+    @property
+    def use_revisions_as_document_ids(self) -> Optional[bool]:
+        return self._data.get("usesRevisionsAsDocumentIds")
+
+    @property
+    def sync_by_revision(self) -> Optional[bool]:
+        return self._data.get("syncByRevision")
+
+    @property
+    def status_string(self) -> Optional[str]:
+        return self._data.get("statusString")
+
+    @property
+    def id(self) -> str:
+        return self._data["id"]  # type: ignore[no-any-return]
+
+    @property
+    def is_smart_child(self) -> bool:
+        return self._data["isSmartChild"]  # type: ignore[no-any-return]
+
+    @property
+    def schema(self) -> Optional[Json]:
+        return self._data.get("schema")
+
+    @property
+    def name(self) -> str:
+        return self._data["name"]  # type: ignore[no-any-return]
+
+    @property
+    def type(self) -> CollectionType:
+        return CollectionType.from_int(self._data["type"])
+
+    @property
+    def status(self) -> CollectionStatus:
+        return CollectionStatus.from_int(self._data["status"])
+
+    @property
+    def count(self) -> int:
+        return self._data["count"]  # type: ignore[no-any-return]
+
+    @property
+    def cache_enabled(self) -> Optional[bool]:
+        return self._data.get("cacheEnabled")
+
+    @property
+    def is_system(self) -> bool:
+        return self._data["isSystem"]  # type: ignore[no-any-return]
+
+    @property
+    def internal_validator_type(self) -> Optional[int]:
+        return self._data.get("internalValidatorType")
+
+    @property
+    def globally_unique_id(self) -> str:
+        return self._data["globallyUniqueId"]  # type: ignore[no-any-return]
+
+    @property
+    def key_options(self) -> KeyOptions:
+        return KeyOptions(self._data["keyOptions"])
+
+    @property
+    def computed_values(self) -> Optional[Json]:
+        return self._data.get("computedValues")
+
+    @property
+    def object_id(self) -> str:
+        return self._data["objectId"]  # type: ignore[no-any-return]
+
+
 class IndexProperties(JsonWrapper):
     """Properties of an index.
 

tests/test_collection.py

@@ -5,6 +5,7 @@
 from arangoasync.errno import DATA_SOURCE_NOT_FOUND, INDEX_NOT_FOUND
 from arangoasync.exceptions import (
     CollectionPropertiesError,
+    CollectionStatisticsError,
     CollectionTruncateError,
     DocumentCountError,
     IndexCreateError,
@@ -30,6 +31,11 @@ async def test_collection_misc_methods(doc_col, bad_col):
     assert len(properties.format()) > 1
     with pytest.raises(CollectionPropertiesError):
         await bad_col.properties()
+    statistics = await doc_col.statistics()
+    assert statistics.name == doc_col.name
+    assert "figures" in statistics
+    with pytest.raises(CollectionStatisticsError):
+        await bad_col.statistics()
 
 
 @pytest.mark.asyncio

tests/test_typings.py

@@ -2,6 +2,7 @@
 
 from arangoasync.typings import (
     CollectionInfo,
+    CollectionStatistics,
     CollectionStatus,
     CollectionType,
     EdgeDefinitionOptions,
@@ -386,3 +387,62 @@ def test_EdgeDefinitionOptions():
     )
 
     assert options.satellites == ["col1", "col2"]
+
+
+def test_CollectionStatistics():
+    data = {
+        "figures": {
+            "indexes": {"count": 1, "size": 1234},
+            "documentsSize": 5601,
+            "cacheInUse": False,
+            "cacheSize": 0,
+            "cacheUsage": 0,
+        },
+        "writeConcern": 1,
+        "waitForSync": False,
+        "usesRevisionsAsDocumentIds": True,
+        "syncByRevision": True,
+        "statusString": "loaded",
+        "id": "69123",
+        "isSmartChild": False,
+        "schema": None,
+        "name": "products",
+        "type": 2,
+        "status": 3,
+        "count": 1,
+        "cacheEnabled": False,
+        "isSystem": False,
+        "internalValidatorType": 0,
+        "globallyUniqueId": "hB7C02EE43DCE/69123",
+        "keyOptions": {
+            "allowUserKeys": True,
+            "type": "traditional",
+            "lastValue": 69129,
+        },
+        "computedValues": None,
+        "objectId": "69124",
+    }
+
+    stats = CollectionStatistics(data)
+
+    assert stats.figures == data["figures"]
+    assert stats.write_concern == 1
+    assert stats.wait_for_sync is False
+    assert stats.use_revisions_as_document_ids is True
+    assert stats.sync_by_revision is True
+    assert stats.status_string == "loaded"
+    assert stats.id == "69123"
+    assert stats.is_smart_child is False
+    assert stats.schema is None
+    assert stats.name == "products"
+    assert stats.type == CollectionType.DOCUMENT
+    assert stats.status == CollectionStatus.LOADED
+    assert stats.count == 1
+    assert stats.cache_enabled is False
+    assert stats.is_system is False
+    assert stats.internal_validator_type == 0
+    assert stats.globally_unique_id == "hB7C02EE43DCE/69123"
+    assert isinstance(stats.key_options, KeyOptions)
+    assert stats.key_options["type"] == "traditional"
+    assert stats.computed_values is None
+    assert stats.object_id == "69124"