new: server_maintenance_mode and toggle_server_maintenance_mode

Author: aMahanna

arango/cluster.py

@@ -140,6 +140,58 @@ def response_handler(resp: Response) -> Json:
 
         return self._execute(request, response_handler)
 
+    def server_maintenance_mode(self, server_id: str) -> Result[Json]:
+        """Return the maintenance status for the given server.
+
+        :param server_id: Server ID.
+        :type server_id: str
+        :return: Maintenance status for the given server.
+        :rtype: dict
+        :raise arango.exceptions.ClusterMaintenanceModeError: If retrieval fails.
+        """
+        request = Request(
+            method="get",
+            endpoint=f"/_admin/cluster/maintenance/{server_id}",
+        )
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                result: Json = resp.body.get("result", {})
+                return result
+
+            raise ClusterMaintenanceModeError(resp, request)
+
+        return self._execute(request, response_handler)
+
+    def toggle_server_maintenance_mode(
+        self, server_id: str, mode: str, timeout: Optional[int] = None
+    ) -> Result[Json]:
+        """Enable or disable the maintenance mode for the given server.
+
+        :param server_id: Server ID.
+        :type server_id: str
+        :param mode: Maintenance mode. Allowed values are "normal" and "maintenance".
+        :type mode: str
+        :param timeout: Timeout in seconds.
+        :type timeout: Optional[int]
+        :return: Result of the operation.
+        :rtype: dict
+        :raise arango.exceptions.ClusterMaintenanceModeError: If toggle fails.
+        """
+        request = Request(
+            method="put",
+            endpoint=f"/_admin/cluster/maintenance/{server_id}",
+            data={"mode": mode, "timeout": timeout},
+        )
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                return format_body(resp.body)
+
+            raise ClusterMaintenanceModeError(resp, request)
+
+        return self._execute(request, response_handler)
+
     def health(self) -> Result[Json]:
         """Return the cluster health.
 

docs/cluster.rst

@@ -86,8 +86,13 @@ Below is an example on how to manage clusters using python-arango.
     cluster.server_engine(server_id)
     cluster.server_version(server_id)
     cluster.server_statistics(server_id)
+    cluster.server_maintenance_mode(server_id)
 
-    # Toggle maintenance mode (allowed values are "on" and "off").
+    # Toggle Server maintenance mode (allowed values are "normal" and "maintenance").
+    cluster.toggle_server_maintenance_mode(server_id, 'normal')
+    cluster.toggle_server_maintenance_mode(server_id, 'maintenance', timeout=30)
+
+    # Toggle Cluster maintenance mode (allowed values are "on" and "off").
     cluster.toggle_maintenance_mode('on')
     cluster.toggle_maintenance_mode('off')
 

tests/test_cluster.py

@@ -1,5 +1,6 @@
 import warnings
 
+import time
 import pytest
 from packaging import version
 
@@ -99,6 +100,28 @@ def test_cluster_server_statistics(sys_db, bad_db, cluster):
     assert err.value.error_code in {FORBIDDEN, DATABASE_NOT_FOUND}
 
 
+def test_cluster_server_maintenance_mode(sys_db, bad_db, cluster):
+    if not cluster:
+        pytest.skip("Only tested in a cluster setup")
+
+    server_id = sys_db.cluster.server_id()
+    result = sys_db.cluster.server_maintenance_mode(server_id)
+    assert result == {}
+
+    with assert_raises(ClusterMaintenanceModeError) as err:
+        bad_db.cluster.server_maintenance_mode(server_id)
+    assert err.value.error_code in {FORBIDDEN, DATABASE_NOT_FOUND}
+
+    sys_db.cluster.toggle_server_maintenance_mode(server_id, "maintenance", timeout=1)
+    result = sys_db.cluster.server_maintenance_mode(server_id)
+    assert "Mode" in result
+    assert "Until" in result
+
+    time.sleep(1)
+    result = sys_db.cluster.server_maintenance_mode(server_id)
+    assert result == {}
+
+
 def test_cluster_toggle_maintenance_mode(sys_db, bad_db, cluster):
     if not cluster:
         pytest.skip("Only tested in a cluster setup")