Merge pull request #18 from wherobots/max/cancel-query

Support query cancellation

Author: mpetazzoni

poetry.lock

@@ -14,7 +14,7 @@ Tracker = "https://github.com/wherobots/wherobots-python-dbapi-driver/issues"
 [tool.poetry.dependencies]
 python = "^3.9"
 requests = "^2.31.0"
-websockets = "^12.0"
+websockets = "13.0"
 tenacity = "^8.2.3"
 pyarrow = "^14.0.2"
 cbor2 = "^5.6.3"

pyproject.toml

@@ -131,6 +131,9 @@ def __listen(self):
             # Incoming state transitions are handled here.
             if query.state == ExecutionState.SUCCEEDED:
                 self.__request_results(execution_id)
+            elif query.state == ExecutionState.CANCELLED:
+                logging.info("Query %s has been cancelled.", execution_id)
+                self.__queries.pop(execution_id)
             elif query.state == ExecutionState.FAILED:
                 # Don't do anything here; the ERROR event is coming with more
                 # details.
@@ -230,7 +233,14 @@ def __request_results(self, execution_id: str) -> None:
         self.__send(request)
 
     def __cancel_query(self, execution_id: str) -> None:
-        query = self.__queries.pop(execution_id)
-        if query:
-            logging.info("Cancelled query %s.", execution_id)
-            # TODO: when protocol supports it, send cancellation request.
+        """Cancels the query with the given execution ID."""
+        query = self.__queries.get(execution_id)
+        if not query:
+            return
+
+        request = {
+            "kind": RequestKind.CANCEL.value,
+            "execution_id": execution_id,
+        }
+        logging.info("Cancelling query %s...", execution_id)
+        self.__send(request)

wherobots/db/connection.py

@@ -1,4 +1,5 @@
 from enum import auto
+from packaging.version import Version
 from strenum import LowercaseStrEnum, StrEnum
 
 from .region import Region
@@ -15,7 +16,7 @@
 DEFAULT_REUSE_SESSION: bool = True
 
 MAX_MESSAGE_SIZE: int = 100 * 2**20  # 100MiB
-PROTOCOL_VERSION: str = "1.0.0"
+PROTOCOL_VERSION: Version = Version("1.0.0")
 
 
 class ExecutionState(LowercaseStrEnum):
@@ -31,6 +32,9 @@ class ExecutionState(LowercaseStrEnum):
     SUCCEEDED = auto()
     "The SQL session has reported the query has completed successfully."
 
+    CANCELLED = auto()
+    "The SQL session has reported the query has been cancelled."
+
     FAILED = auto()
     "The SQL session has reported the query has failed."
 
@@ -41,12 +45,17 @@ class ExecutionState(LowercaseStrEnum):
     "The driver has completed processing the query results."
 
     def is_terminal_state(self):
-        return self in (ExecutionState.COMPLETED, ExecutionState.FAILED)
+        return self in (
+            ExecutionState.COMPLETED,
+            ExecutionState.CANCELLED,
+            ExecutionState.FAILED,
+        )
 
 
 class RequestKind(LowercaseStrEnum):
     EXECUTE_SQL = auto()
     RETRIEVE_RESULTS = auto()
+    CANCEL = auto()
 
 
 class EventKind(LowercaseStrEnum):

wherobots/db/constants.py

@@ -89,7 +89,7 @@ def executemany(self, operation: str, seq_of_parameters: list[dict[str, Any]]):
 
     def fetchone(self):
         results = self.__get_results()[self.__current_row :]
-        if not results:
+        if len(results) == 0:
             return None
         self.__current_row += 1
         return results[0]
@@ -105,7 +105,8 @@ def fetchall(self):
 
     def close(self):
         """Close the cursor."""
-        pass
+        if self.__results is None and self.__current_execution_id:
+            self.__cancel_fn(self.__current_execution_id)
 
     def __iter__(self):
         return self

wherobots/db/cursor.py

@@ -3,19 +3,19 @@
 A PEP-0249 compatible driver for interfacing with Wherobots DB.
 """
 
+from importlib import metadata
+from importlib.metadata import PackageNotFoundError
 import logging
+from packaging.version import Version
 import platform
-import urllib.parse
 import queue
-from importlib import metadata
-from importlib.metadata import PackageNotFoundError
-
 import requests
 import tenacity
-import threading
 from typing import Union
+import urllib.parse
 import websockets.sync.client
 
+from .connection import Connection
 from .constants import (
     DEFAULT_ENDPOINT,
     DEFAULT_REGION,
@@ -36,7 +36,6 @@
 )
 from .region import Region
 from .runtime import Runtime
-from .connection import Connection
 
 apilevel = "2.0"
 threadsafety = 1
@@ -163,41 +162,28 @@ def http_to_ws(uri: str) -> str:
 
 def connect_direct(
     uri: str,
+    protocol: Version = PROTOCOL_VERSION,
     headers: dict[str, str] = None,
     read_timeout: float = DEFAULT_READ_TIMEOUT_SECONDS,
     results_format: Union[ResultsFormat, None] = None,
     data_compression: Union[DataCompression, None] = None,
     geometry_representation: Union[GeometryRepresentation, None] = None,
 ) -> Connection:
     q = queue.SimpleQueue()
-    uri_with_protocol = f"{uri}/{PROTOCOL_VERSION}"
-
-    def create_ws_connection():
-        try:
-            logging.info("Connecting to SQL session at %s ...", uri_with_protocol)
-            ws = websockets.sync.client.connect(
-                uri=uri_with_protocol,
-                additional_headers=headers,
-                max_size=MAX_MESSAGE_SIZE,
-            )
-            q.put(ws)
-        except Exception as e:
-            q.put(e)
-
-    dt = threading.Thread(
-        name="wherobots-ws-connector",
-        target=create_ws_connection,
-        daemon=True,
-    )
-    dt.start()
-    dt.join()
+    uri_with_protocol = f"{uri}/{protocol}"
 
-    result = q.get()
-    if isinstance(result, Exception):
-        raise InterfaceError("Failed to connect to SQL session!") from result
+    try:
+        logging.info("Connecting to SQL session at %s ...", uri_with_protocol)
+        ws = websockets.sync.client.connect(
+            uri=uri_with_protocol,
+            additional_headers=headers,
+            max_size=MAX_MESSAGE_SIZE,
+        )
+    except Exception as e:
+        raise InterfaceError("Failed to connect to SQL session!") from e
 
     return Connection(
-        result,
+        ws,
         read_timeout=read_timeout,
         results_format=results_format,
         data_compression=data_compression,