Implement query rules endpoint for ArangoDB version 3.10 (#216)

Author: tjoubert

arango/aql.py

@@ -19,6 +19,7 @@
     AQLQueryExplainError,
     AQLQueryKillError,
     AQLQueryListError,
+    AQLQueryRulesGetError,
     AQLQueryTrackingGetError,
     AQLQueryTrackingSetError,
     AQLQueryValidateError,
@@ -30,6 +31,7 @@
     format_aql_tracking,
     format_body,
     format_query_cache_entry,
+    format_query_rule_item,
 )
 from arango.request import Request
 from arango.response import Response
@@ -632,3 +634,24 @@ def response_handler(resp: Response) -> Union[bool, Json]:
             return {"deleted": resp.body["deletedCount"]}
 
         return self._execute(request, response_handler)
+
+    def query_rules(self) -> Result[Jsons]:
+        """Return the available optimizer rules for AQL queries
+
+        :return: The available optimizer rules for AQL queries
+        :rtype: dict
+        :raise arango.exceptions.AQLQueryRulesGetError: If retrieval fails.
+        """
+        request = Request(method="get", endpoint="/_api/query/rules")
+
+        def response_handler(resp: Response) -> Jsons:
+            if not resp.is_success:
+                raise AQLQueryRulesGetError(resp, request)
+
+            rules: Jsons = resp.body
+            items: Jsons = []
+            for rule in rules:
+                items.append(format_query_rule_item(rule))
+            return items
+
+        return self._execute(request, response_handler)

arango/exceptions.py

@@ -153,6 +153,10 @@ class AQLFunctionDeleteError(ArangoServerError):
     """Failed to delete AQL user function."""
 
 
+class AQLQueryRulesGetError(ArangoServerError):
+    """Failed to retrieve AQL query rules."""
+
+
 ##############################
 # Async Execution Exceptions #
 ##############################

arango/formatter.py

@@ -1147,3 +1147,47 @@ def format_query_cache_entry(body: Json) -> Json:
         result["data_sources"] = body["dataSources"]
 
     return verify_format(body, result)
+
+
+def format_query_rule_item(body: Json) -> Json:
+    """Format AQL query rule item.
+
+    :param body: Input body.
+    :type body: dict
+    :return: Formatted body.
+    :rtype: dict
+    """
+    result = {}
+
+    if "name" in body:
+        result["name"] = body["name"]
+    if "flags" in body:
+        result["flags"] = format_query_rule_item_flags(body["flags"])
+
+    return verify_format(body, result)
+
+
+def format_query_rule_item_flags(body: Json) -> Json:
+    """Format AQL query rule item flags.
+
+    :param body: Input body.
+    :type body: dict
+    :return: Formatted body.
+    :rtype: dict
+    """
+    result = {}
+
+    if "hidden" in body:
+        result["hidden"] = body["hidden"]
+    if "clusterOnly" in body:
+        result["clusterOnly"] = body["clusterOnly"]
+    if "canBeDisabled" in body:
+        result["canBeDisabled"] = body["canBeDisabled"]
+    if "canCreateAdditionalPlans" in body:
+        result["canCreateAdditionalPlans"] = body["canCreateAdditionalPlans"]
+    if "disabledByDefault" in body:
+        result["disabledByDefault"] = body["disabledByDefault"]
+    if "enterpriseOnly" in body:
+        result["enterpriseOnly"] = body["enterpriseOnly"]
+
+    return verify_format(body, result)