initial commit

Author: aMahanna

arango/database.py

@@ -43,6 +43,8 @@
     ServerEchoError,
     ServerEncryptionError,
     ServerEngineError,
+    ServerLicenseGetError,
+    ServerLicenseSetError,
     ServerLogLevelError,
     ServerLogLevelSetError,
     ServerMetricsError,
@@ -348,6 +350,54 @@ def response_handler(resp: Response) -> Json:
 
         return self._execute(request, response_handler)
 
+    def license(self) -> Result[Json]:
+        """View the license information and status of an
+        Enterprise Edition instance. Can be called on
+        single servers, Coordinators, and DB-Servers.
+
+        :return: Server license.
+        :rtype: dict
+        :raise arango.exceptions.ServerLicenseGetError: If retrieval fails.
+        """
+        request = Request(method="get", endpoint="/_admin/license")
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                result: Json = resp.body
+                return result
+            raise ServerLicenseGetError(resp, request)
+
+        return self._execute(request, response_handler)
+
+    def set_license(self, license: str, force: bool = False) -> Result[Json]:
+        """Set a new license for an Enterprise Edition
+        instance. Can be called on single servers, Coordinators,
+        and DB-Servers.
+
+        :param license: The Base64-encoded license string.
+        :type license: str
+        :param force: If set to True, the new license will be set even if
+            it expires sooner than the current license.
+        :type force: bool
+        :return: Server license.
+        :rtype: dict
+        :raise arango.exceptions.ServerLicenseError: If retrieval fails.
+        """
+        request = Request(
+            method="put",
+            endpoint="/_admin/license",
+            params={"force": force},
+            data=license,
+        )
+
+        def response_handler(resp: Response) -> Json:
+            if resp.is_success:
+                result: Json = resp.body
+                return result
+            raise ServerLicenseSetError(resp, request)
+
+        return self._execute(request, response_handler)
+
     def status(self) -> Result[Json]:
         """Return ArangoDB server status.
 

arango/exceptions.py

@@ -614,6 +614,14 @@ class ServerDetailsError(ArangoServerError):
     """Failed to retrieve server details."""
 
 
+class ServerLicenseGetError(ArangoServerError):
+    """Failed to retrieve server license."""
+
+
+class ServerLicenseSetError(ArangoServerError):
+    """Failed to set server license."""
+
+
 class ServerStatusError(ArangoServerError):
     """Failed to retrieve server status."""
 

tests/test_database.py

@@ -319,3 +319,20 @@ def test_database_utf8(sys_db, db_version, special_db_names):
         assert sys_db.create_database(name)
         assert sys_db.has_database(name)
         assert sys_db.delete_database(name)
+
+
+def test_license(sys_db, db, enterprise):
+    license = sys_db.license()
+    assert isinstance(license, dict)
+
+    if enterprise:
+        assert set(license.keys()) == {
+            "upgrading",
+            "features",
+            "hash",
+            "license",
+            "version",
+            "status",
+        }
+    else:
+        assert license == {"license": "none"}