PYCO-94: Support Server Side Async Requests

Changes
-------
* Add start_query() to cluster & scope for both sync and async APIs.

Author: thejcfactor

acouchbase_analytics/cluster.py

@@ -25,7 +25,8 @@
     from typing import TypeAlias
 
 from acouchbase_analytics.database import AsyncDatabase
-from couchbase_analytics.result import AsyncQueryResult
+from acouchbase_analytics.query_handle import AsyncQueryHandle
+from acouchbase_analytics.result import AsyncQueryResult
 
 if TYPE_CHECKING:
     from couchbase_analytics.credential import Credential
@@ -143,6 +144,22 @@ def execute_query(self, statement: str, *args: object, **kwargs: object) -> Awai
         """  # noqa: E501
         return self._impl.execute_query(statement, *args, **kwargs)
 
+    def start_query(self, statement: str, *args: object, **kwargs: object) -> Awaitable[AsyncQueryHandle]:
+        """Executes a query against an Analytics cluster in async mode.
+
+        .. seealso::
+            :meth:`acouchbase_analytics.Scope.start_query`: For how to execute scope-level queries.
+
+        Args:
+            statement: The SQL++ statement to execute.
+            options (:class:`~acouchbase_analytics.options.StartQueryOptions`): Optional parameters for the query operation.
+            **kwargs (Dict[str, Any]): keyword arguments that can be used in place or to override provided :class:`~acouchbase_analytics.options.StartQueryOptions`
+
+        Returns:
+            :class:`~acouchbase_analytics.query_handle.AsyncQueryHandle`: An instance of a :class:`~acouchbase_analytics.query_handle.AsyncQueryHandle`
+        """  # noqa: E501
+        return self._impl.start_query(statement, *args, **kwargs)
+
     async def shutdown(self) -> None:
         """Shuts down this cluster instance. Cleaning up all resources associated with it.
 

acouchbase_analytics/cluster.pyi

@@ -21,10 +21,19 @@ if sys.version_info < (3, 11):
 else:
     from typing import Unpack
 
+from acouchbase_analytics import JSONType
+from acouchbase_analytics.credential import Credential
 from acouchbase_analytics.database import AsyncDatabase
-from couchbase_analytics.credential import Credential
-from couchbase_analytics.options import ClusterOptions, ClusterOptionsKwargs, QueryOptions, QueryOptionsKwargs
-from couchbase_analytics.result import AsyncQueryResult
+from acouchbase_analytics.options import (
+    ClusterOptions,
+    ClusterOptionsKwargs,
+    QueryOptions,
+    QueryOptionsKwargs,
+    StartQueryOptions,
+    StartQueryOptionsKwargs,
+)
+from acouchbase_analytics.query_handle import AsyncQueryHandle
+from acouchbase_analytics.result import AsyncQueryResult
 
 class AsyncCluster:
     @overload
@@ -54,14 +63,34 @@ class AsyncCluster:
     ) -> Awaitable[AsyncQueryResult]: ...
     @overload
     def execute_query(
-        self, statement: str, options: QueryOptions, *args: str, **kwargs: Unpack[QueryOptionsKwargs]
+        self, statement: str, options: QueryOptions, *args: JSONType, **kwargs: Unpack[QueryOptionsKwargs]
     ) -> Awaitable[AsyncQueryResult]: ...
     @overload
     def execute_query(
-        self, statement: str, options: QueryOptions, *args: str, **kwargs: str
+        self, statement: str, options: QueryOptions, *args: JSONType, **kwargs: str
     ) -> Awaitable[AsyncQueryResult]: ...
     @overload
-    def execute_query(self, statement: str, *args: str, **kwargs: str) -> Awaitable[AsyncQueryResult]: ...
+    def execute_query(self, statement: str, *args: JSONType, **kwargs: str) -> Awaitable[AsyncQueryResult]: ...
+    @overload
+    def start_query(self, statement: str) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(self, statement: str, options: StartQueryOptions) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(self, statement: str, **kwargs: Unpack[StartQueryOptionsKwargs]) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(
+        self, statement: str, options: StartQueryOptions, **kwargs: Unpack[StartQueryOptionsKwargs]
+    ) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(
+        self, statement: str, options: StartQueryOptions, *args: JSONType, **kwargs: Unpack[StartQueryOptionsKwargs]
+    ) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(
+        self, statement: str, options: StartQueryOptions, *args: JSONType, **kwargs: str
+    ) -> AsyncQueryHandle: ...
+    @overload
+    def start_query(self, statement: str, *args: JSONType, **kwargs: str) -> AsyncQueryHandle: ...
     def shutdown(self) -> Awaitable[None]: ...
     @overload
     @classmethod

acouchbase_analytics/options.py

@@ -16,9 +16,13 @@
 
 from couchbase_analytics.common.options import ClusterOptions as ClusterOptions  # noqa: F401
 from couchbase_analytics.common.options import ClusterOptionsKwargs as ClusterOptionsKwargs  # noqa: F401
+from couchbase_analytics.common.options import FetchResultsOptions as FetchResultsOptions  # noqa: F401
+from couchbase_analytics.common.options import FetchResultsOptionsKwargs as FetchResultsOptionsKwargs  # noqa: F401
 from couchbase_analytics.common.options import QueryOptions as QueryOptions  # noqa: F401
 from couchbase_analytics.common.options import QueryOptionsKwargs as QueryOptionsKwargs  # noqa: F401
 from couchbase_analytics.common.options import SecurityOptions as SecurityOptions  # noqa: F401
 from couchbase_analytics.common.options import SecurityOptionsKwargs as SecurityOptionsKwargs  # noqa: F401
+from couchbase_analytics.common.options import StartQueryOptions as StartQueryOptions  # noqa: F401
+from couchbase_analytics.common.options import StartQueryOptionsKwargs as StartQueryOptionsKwargs  # noqa: F401
 from couchbase_analytics.common.options import TimeoutOptions as TimeoutOptions  # noqa: F401
 from couchbase_analytics.common.options import TimeoutOptionsKwargs as TimeoutOptionsKwargs  # noqa: F401

acouchbase_analytics/protocol/_core/client_adapter.py

@@ -17,20 +17,18 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Optional, cast
+from typing import Optional, Union, cast
 from uuid import uuid4
 
 from httpx import URL, AsyncClient, BasicAuth, Response
 
 from couchbase_analytics.common.credential import Credential
 from couchbase_analytics.common.deserializer import Deserializer
 from couchbase_analytics.common.logging import LogLevel, log_message
+from couchbase_analytics.protocol._core.request import CancelRequest, HttpRequest, QueryRequest, StartQueryRequest
 from couchbase_analytics.protocol.connection import _ConnectionDetails
 from couchbase_analytics.protocol.options import OptionsBuilder
 
-if TYPE_CHECKING:
-    from couchbase_analytics.protocol._core.request import QueryRequest
-
 
 class _AsyncClientAdapter:
     """
@@ -164,7 +162,9 @@ async def create_client(self) -> None:
     def log_message(self, message: str, log_level: LogLevel) -> None:
         log_message(logger, f'{self.log_prefix} {message}', log_level)
 
-    async def send_request(self, request: QueryRequest) -> Response:
+    async def send_request(
+        self, request: Union[CancelRequest, HttpRequest, QueryRequest, StartQueryRequest], stream: Optional[bool] = True
+    ) -> Response:
         """
         **INTERNAL**
         """
@@ -177,8 +177,19 @@ async def send_request(self, request: QueryRequest) -> Response:
             port=request.url.port,
             path=request.url.path,
         )
-        req = self._client.build_request(request.method, url, json=request.body, extensions=request.extensions)
-        return await self._client.send(req, stream=True)
+        if isinstance(request, (QueryRequest, StartQueryRequest)):
+            req = self._client.build_request(request.method, url, json=request.body, extensions=request.extensions)
+        else:
+            headers = request.headers if request.headers is not None else None
+            data = request.data if isinstance(request, CancelRequest) else None
+            req = self._client.build_request(
+                request.method, url, data=data, headers=headers, extensions=request.extensions
+            )
+
+        if stream is None:
+            stream = True
+
+        return await self._client.send(req, stream=stream)
 
     def reset_client(self) -> None:
         """