arangodb · aMahanna · Jan 26, 2024 · Dec 29, 2023 · Dec 29, 2023 · Dec 29, 2023
diff --git a/arango/database.py b/arango/database.py
@@ -27,6 +27,7 @@
     CollectionCreateError,
     CollectionDeleteError,
     CollectionListError,
+    DatabaseCompactError,
     DatabaseCreateError,
     DatabaseDeleteError,
     DatabaseListError,
@@ -439,6 +440,47 @@ def response_handler(resp: Response) -> Json:
 
         return self._execute(request, response_handler)
 
+    def compact(
+        self,
+        change_level: Optional[bool] = None,
+        compact_bottom_most_level: Optional[bool] = None,
+    ) -> Result[Json]:
+        """Compact all databases.
+
+        NOTE: This command can cause a full rewrite of all data in all databases,
+        which may take very long for large databases. It should thus only be used with
+        care and only when additional I/O load can be tolerated for a prolonged time.
+
+        This method can be used to reclaim disk space after substantial data deletions
+        have taken place, by compacting the entire database system data.
+
+        This method requires superuser access.
+
        :param change_level: Whether or not compacted data should be moved to
+            the minimum possible level. Default value is False.
+        :type change_level: bool | None
+        :param compact_bottom_most_level: Whether or not to compact the
+            bottom-most level of data. Default value is False.
+        :type compact_bottom_most_level: bool | None
+        :return: Collection compact.
+        :rtype: dict
+        :raise arango.exceptions.CollectionCompactError: If retrieval fails.
+        """
+        data = {}
+        if change_level is not None:
+            data["changeLevel"] = change_level
+        if compact_bottom_most_level is not None:
+            data["compactBottomMostLevel"] = compact_bottom_most_level
+
+        request = Request(method="put", endpoint="/_admin/compact", data=data)
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                return format_body(resp.body)
+            raise DatabaseCompactError(resp, request)
+
+        return self._execute(request, response_handler)
+
     def required_db_version(self) -> Result[str]:
         """Return required version of target database.
 

diff --git a/arango/exceptions.py b/arango/exceptions.py
@@ -360,6 +360,10 @@ class DatabaseDeleteError(ArangoServerError):
     """Failed to delete database."""
 
 
+class DatabaseCompactError(ArangoServerError):
+    """Failed to compact databases."""
+
+
 #######################
 # Document Exceptions #
 #######################

diff --git a/tests/test_database.py b/tests/test_database.py
@@ -13,6 +13,7 @@
     USE_SYSTEM_DATABASE,
 )
 from arango.exceptions import (
+    DatabaseCompactError,
     DatabaseCreateError,
     DatabaseDeleteError,
     DatabaseListError,
@@ -37,7 +38,12 @@
 from arango.pregel import Pregel
 from arango.replication import Replication
 from arango.wal import WAL
-from tests.helpers import assert_raises, generate_db_name
+from tests.helpers import (
+    assert_raises,
+    generate_col_name,
+    generate_db_name,
+    generate_jwt,
+)
 
 
 def test_database_attributes(db, username):
@@ -57,7 +63,7 @@ def test_database_attributes(db, username):
     assert isinstance(db.wal, WAL)
 
 
-def test_database_misc_methods(sys_db, db, bad_db, cluster):
+def test_database_misc_methods(client, sys_db, db, bad_db, cluster, secret):
     # Test get properties
     properties = db.properties()
     assert "id" in properties
@@ -263,6 +269,18 @@ def test_database_misc_methods(sys_db, db, bad_db, cluster):
         bad_db.engine()
     assert err.value.error_code in {11, 1228}
 
+    # Test database compact
+    with assert_raises(DatabaseCompactError) as err:
+        db.compact()
+
+    collection = db.create_collection(generate_col_name())
+    collection.insert({"foo": "bar"})
+
+    token = generate_jwt(secret)
+    db_superuser = client.db(db.name, superuser_token=token)
+    result = db_superuser.compact()
+    assert result == {}
+
 
 def test_database_management(db, sys_db, bad_db):
     # Test list databases