Adding QueueBoundedDatabase along with the appropriate executor.

Author: apetenchea

arango/database.py

@@ -1,4 +1,10 @@
-__all__ = ["StandardDatabase", "AsyncDatabase", "BatchDatabase", "TransactionDatabase"]
+__all__ = [
+    "StandardDatabase",
+    "AsyncDatabase",
+    "BatchDatabase",
+    "QueueBoundedDatabase",
+    "TransactionDatabase",
+]
 
 from datetime import datetime
 from numbers import Number
@@ -75,6 +81,7 @@
     AsyncApiExecutor,
     BatchApiExecutor,
     DefaultApiExecutor,
+    QueueBoundedApiExecutor,
     TransactionApiExecutor,
 )
 from arango.formatter import (
@@ -2514,6 +2521,19 @@ def begin_transaction(
             max_size=max_size,
         )
 
+    def begin_queue_bounded_execution(
+        self, max_queue_time_seconds: float = 0.0
+    ) -> "QueueBoundedDatabase":
+        """Begin queue bounded execution.
+
+        :param max_queue_time_seconds: Maximum time in seconds a request can be queued
+            on the server-side. If set to 0, the server ignores this setting.
+        :type max_queue_time_seconds: float
+        :return: Database API wrapper object specifically for queue bounded execution.
+        :rtype: arango.database.QueueBoundedDatabase
+        """
+        return QueueBoundedDatabase(self._conn, max_queue_time_seconds)
+
 
 class AsyncDatabase(Database):
     """Database API wrapper tailored specifically for async execution.
@@ -2684,3 +2704,44 @@ def abort_transaction(self) -> bool:
         :raise arango.exceptions.TransactionAbortError: If abort fails.
         """
         return self._executor.abort()
+
+
+class QueueBoundedDatabase(Database):
+    """Database API wrapper tailored specifically to hande time-bound requests.
+
+    See :func:`arango.database.StandardDatabase.begin_queue_bounded_execution`.
+
+    :param connection: HTTP connection.
+    :param max_queue_time_seconds: Maximum server-side queuing time in seconds.
+        If the server-side queuing time exceeds the client's specified limit,
+        the request will be rejected.
+    :type max_queue_time_seconds: float
+    """
+
+    def __init__(self, connection: Connection, max_queue_time_seconds: float) -> None:
+        self._executor: QueueBoundedApiExecutor
+        super().__init__(
+            connection=connection,
+            executor=QueueBoundedApiExecutor(connection, max_queue_time_seconds),
+        )
+
+    def __repr__(self) -> str:
+        return f"<QueueBoundedDatabase {self.name}>"
+
+    @property
+    def last_queue_time(self) -> float:
+        """Return the most recently recorded server-side queuing time in seconds.
+
+        :return: Server-side queuing time in seconds.
+        :rtype: float
+        """
+        return self._executor.queue_time_seconds
+
+    def adjust_max_queue_time(self, max_queue_time_seconds: float) -> None:
+        """Adjust the maximum server-side queuing time in seconds.
+
+        :param max_queue_time_seconds: New maximum server-side queuing time
+            in seconds.
+        :type max_queue_time_seconds: float
+        """
+        self._executor.max_queue_time_seconds = max_queue_time_seconds

arango/exceptions.py

@@ -232,6 +232,15 @@ class BatchExecuteError(ArangoServerError):
     """Failed to execute batch API request."""
 
 
+######################################
+# Queue Bounded Execution Exceptions #
+######################################
+
+
+class QueueBoundedExecutorError(ArangoServerError):
+    """Failed to execute queue-bounded API request."""
+
+
 #########################
 # Collection Exceptions #
 #########################

arango/executor.py

@@ -4,7 +4,7 @@
     "AsyncApiExecutor",
     "BatchApiExecutor",
     "TransactionApiExecutor",
-    "QueueTimeApiExecutor",
+    "QueueBoundedApiExecutor",
 ]
 
 from collections import OrderedDict
@@ -17,6 +17,7 @@
     AsyncExecuteError,
     BatchExecuteError,
     BatchStateError,
+    QueueBoundedExecutorError,
     TransactionAbortError,
     TransactionCommitError,
     TransactionInitError,
@@ -33,7 +34,7 @@
     "AsyncApiExecutor",
     "BatchApiExecutor",
     "TransactionApiExecutor",
-    "QueueTimeApiExecutor",
+    "QueueBoundedApiExecutor",
 ]
 
 T = TypeVar("T")
@@ -432,26 +433,57 @@ def abort(self) -> bool:
         raise TransactionAbortError(resp, request)
 
 
-class QueueTimeApiExecutor:
-    """API executor that handles queue time.
+class QueueBoundedApiExecutor:
+    """Allows setting the maximum acceptable server-side queuing time
+        for client requests.
 
     :param connection: HTTP connection.
     :type connection: arango.connection.BasicConnection |
         arango.connection.JwtConnection | arango.connection.JwtSuperuserConnection
+    :param max_queue_time_seconds: Maximum server-side queuing time in seconds.
+    :type max_queue_time_seconds: float
     """
 
-    def __init__(self, connection: Connection) -> None:
+    def __init__(self, connection: Connection, max_queue_time_seconds: float) -> None:
         self._conn = connection
+        self._max_queue_time_seconds = max_queue_time_seconds
+        self._queue_time_seconds = 0.0
 
     @property
     def context(self) -> str:
-        return "queue-time"
+        return "queue-bounded"
+
+    @property
+    def queue_time_seconds(self) -> float:
+        """Return the most recent request queuing/de-queuing time.
+
+        :return: Server-side queuing time in seconds.
+        :rtype: float
+        """
+        return self._queue_time_seconds
+
+    @property
+    def max_queue_time_seconds(self) -> float:
+        """Return the maximum server-side queuing time.
+
+        :return: Maximum server-side queuing time in seconds.
+        :rtype: float
+        """
+        return self._max_queue_time_seconds
+
+    @max_queue_time_seconds.setter
+    def max_queue_time_seconds(self, value: float) -> None:
+        """Set the maximum server-side queuing time.
+
+        :param value: Maximum server-side queuing time in seconds.
+        :type value: float
+        """
+        self._max_queue_time_seconds = value
 
     def execute(
         self,
         request: Request,
         response_handler: Callable[[Response], T],
-        max_queue_time_seconds: int,
     ) -> T:
         """Execute an API request and return the result.
 
@@ -461,8 +493,17 @@ def execute(
         :type response_handler: callable
         :return: API execution result.
         """
-        request.headers["x-arango-max-queue-time-seconds"] = max_queue_time_seconds
+        request.headers["x-arango-queue-time-seconds"] = str(
+            self._max_queue_time_seconds
+        )
         resp = self._conn.send_request(request)
 
-        queue_time_seconds = resp.headers["x-arango-queue-time-seconds"]
-        return response_handler(resp, queue_time_seconds)
+        if not resp.is_success:
+            raise QueueBoundedExecutorError(resp, request)
+
+        if "X-Arango-Queue-Time-Seconds" in resp.headers:
+            self._queue_time_seconds = float(
+                resp.headers["X-Arango-Queue-Time-Seconds"]
+            )
+
+        return response_handler(resp)