Updates to move baseline SDK to internal preview ready.

Changes
=======
* Introduce request_context to handle bulk of logic for a request
  (state, retries, etc.)
* Increase test coverage
* Apply formatting/lint via ruff
* Apply static type checking via mypy
* Cleanup and reorganize code structure

Author: thejcfactor

MANIFEST.in

@@ -3,6 +3,8 @@ include couchbase-sdk-analytics-python-black-duck-manifest.yaml
 include couchbase_analytics/common/core/_nonprod_certificates/*.pem
 include couchbase_analytics/common/core/_capella_certificates/*.pem
 recursive-include couchbase_analytics *.py
+exclude couchbase_analytics/tests/*.py
 recursive-include acouchbase_analytics *.py
+exclude acouchbase_analytics/tests/*.py
 global-exclude *.py[cod] *.DS_Store
 exclude .git .gitignore .gitmodules gocaves* *.jar .clang* .cmake* .pre* .flake* MANIFEST.in

acouchbase_analytics/cluster.py

@@ -16,7 +16,7 @@
 from __future__ import annotations
 
 import sys
-from typing import Awaitable, TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Awaitable, Optional
 
 if sys.version_info < (3, 10):
     from typing_extensions import TypeAlias

acouchbase_analytics/cluster.pyi

@@ -23,10 +23,7 @@ else:
 
 from acouchbase_analytics.database import AsyncDatabase
 from couchbase_analytics.credential import Credential
-from couchbase_analytics.options import (ClusterOptions,
-                                         ClusterOptionsKwargs,
-                                         QueryOptions,
-                                         QueryOptionsKwargs)
+from couchbase_analytics.options import ClusterOptions, ClusterOptionsKwargs, QueryOptions, QueryOptionsKwargs
 from couchbase_analytics.result import AsyncQueryResult
 
 class AsyncCluster:

…base_analytics/protocol/core/__init__.py …ase_analytics/protocol/_core/__init__.pyacouchbase_analytics/protocol/core/__init__.py renamed to acouchbase_analytics/protocol/_core/__init__.py acouchbase_analytics/protocol/core/__init__.py renamed to acouchbase_analytics/protocol/_core/__init__.py

@@ -5,12 +5,16 @@
 
 import anyio
 
+
 def get_time() -> float:
     """
     Get the current time in seconds since the epoch.
     """
     return anyio.current_time()
 
+async def sleep(delay: float) -> None:
+    await anyio.sleep(delay)
+
 class AsyncBackend:
     def __init__(self, backend_lib: str) -> None:
         """

…_analytics/protocol/core/_anyio_utils.py …_analytics/protocol/_core/anyio_utils.pyacouchbase_analytics/protocol/core/_anyio_utils.py renamed to acouchbase_analytics/protocol/_core/anyio_utils.py acouchbase_analytics/protocol/core/_anyio_utils.py renamed to acouchbase_analytics/protocol/_core/anyio_utils.py

@@ -15,22 +15,21 @@
 
 from __future__ import annotations
 
-from typing import (AsyncIterator,
-                    Optional)
+from typing import AsyncIterator, Optional
 
 import ijson
-from anyio import (create_memory_object_stream,
-                   Event,
-                   EndOfStream)
-
-
-from couchbase_analytics.common.core.async_json_token_parser import AsyncJsonTokenParser
-from couchbase_analytics.common.core.json_parsing import (JsonParsingError,
-                                                          JsonStreamConfig,
-                                                          ParsedResult,
-                                                          ParsedResultType)
+from anyio import EndOfStream, Event, create_memory_object_stream
+
+from acouchbase_analytics.protocol._core.async_json_token_parser import AsyncJsonTokenParser
+from couchbase_analytics.common._core.json_parsing import (
+    JsonParsingError,
+    JsonStreamConfig,
+    ParsedResult,
+    ParsedResultType,
+)
 from couchbase_analytics.common.errors import AnalyticsError
 
+
 class AsyncJsonStream:
     def __init__(self,
                  http_stream_iter: AsyncIterator[bytes],
@@ -46,14 +45,14 @@ def __init__(self,
         self._http_stream_exhausted = False
 
         # results handling
-        self._send_stream, self._receive_stream = create_memory_object_stream[ParsedResult](max_buffer_size=stream_config.buffered_row_max)
+        self._send_stream, self._receive_stream = create_memory_object_stream[ParsedResult](max_buffer_size=stream_config.buffered_row_max) # noqa: E501
         self._json_stream_parser = None
         self._buffer_entire_result = stream_config.buffer_entire_result
         handler = None if self._buffer_entire_result is True else self._handle_json_result
         self._json_token_parser = AsyncJsonTokenParser(handler)
         self._token_stream_exhausted = False
         self._has_results_or_errors_evt = Event()
-        self._has_results_or_errors_type = ParsedResultType.UNKNOWN
+        self._results_or_errors_type = ParsedResultType.UNKNOWN
 
     @property
     def has_results_or_errors(self) -> Event:
@@ -63,11 +62,11 @@ def has_results_or_errors(self) -> Event:
         return self._has_results_or_errors_evt
 
     @property
-    def has_results_or_errors_type(self) -> ParsedResultType:
+    def results_or_errors_type(self) -> ParsedResultType:
         """
         **INTERNAL**
         """
-        return self._has_results_or_errors_type
+        return self._results_or_errors_type
 
     @property
     def token_stream_exhausted(self) -> bool:
@@ -111,11 +110,11 @@ def _handle_notification(self, result_type: Optional[ParsedResultType]=None) ->
             return
 
         if result_type is None:
-            self._has_results_or_errors_type = ParsedResultType.END
+            self._results_or_errors_type = ParsedResultType.END
             self._has_results_or_errors_evt.set()
             return
 
-        self._has_results_or_errors_type = result_type
+        self._results_or_errors_type = result_type
         self._has_results_or_errors_evt.set()
 
     async def _process_token_stream(self) -> None:
@@ -131,7 +130,7 @@ async def _process_token_stream(self) -> None:
                 # this is a hack b/c the ijson.parse_async iterator does not yield to the event loop
                 # TODO:  create PYCO to either build custom JSON parsing, or dig into ijson root cause
                 await self._json_token_parser.parse_token(event, value)
-            except StopAsyncIteration as ex:
+            except StopAsyncIteration:
                 self._token_stream_exhausted = True
             except ijson.common.IncompleteJSONError as ex:
                 raise JsonParsingError(cause=ex) from None

…alytics/common/core/async_json_stream.py …tics/protocol/_core/async_json_stream.pycouchbase_analytics/common/core/async_json_stream.py renamed to acouchbase_analytics/protocol/_core/async_json_stream.py couchbase_analytics/common/core/async_json_stream.py renamed to acouchbase_analytics/protocol/_core/async_json_stream.py

@@ -15,18 +15,17 @@
 
 from __future__ import annotations
 
-from typing import (Any,
-                    Callable,
-                    Coroutine,
-                    List,
-                    Optional)
+from typing import Any, Callable, Coroutine, List, Optional
+
+from couchbase_analytics.common._core.json_token_parser_base import (
+    POP_EVENTS,
+    START_EVENTS,
+    VALUE_TOKENS,
+    JsonTokenParserBase,
+    ParsingState,
+    TokenType,
+)
 
-from couchbase_analytics.common.core.json_token_parser_base import (JsonTokenParserBase,
-                                                                    ParsingState,
-                                                                    TokenType,
-                                                                    POP_EVENTS,
-                                                                    START_EVENTS,
-                                                                    VALUE_TOKENS,)
 
 class AsyncJsonTokenParser(JsonTokenParserBase):
     def __init__(self,
@@ -37,7 +36,7 @@ def __init__(self,
     async def _handle_obj_emit(self, obj: str) -> bool:
         if (self._emit_results_enabled
             and self._results_handler is not None
-            and self._state == ParsingState.PROCESSING_RESULTS):
+            and ParsingState.okay_to_emit(self._state, self._previous_state)):
             await self._results_handler(bytes(obj, 'utf-8'))
             return True
         return False
@@ -54,13 +53,16 @@ async def _handle_pop_event(self, token_type: TokenType) -> None:
                     obj = f'[{",".join(reversed(obj_pairs))}]'
                 else:
                     obj = f'{{{",".join(reversed(obj_pairs))}}}'
-                object_emitted = await self._handle_obj_emit(obj)
-                if should_emit and object_emitted:
-                    break # this means we emiited the result/error, so stop processing the stack
+            
+                if should_emit:
+                    object_emitted = await self._handle_obj_emit(obj)
+                    if object_emitted:
+                        break # this means we emiited the result/error, so stop processing the stack
 
                 if len(self._stack) > 0 and self._stack[-1].type == TokenType.MAP_KEY:
                     map_key = self._pop()
-                    # If we are emitting rows and/or errors, we don't keep them in the stack and therefore don't need to return the results           
+                    # If we are emitting rows and/or errors, 
+                    # we don't keep them in the stack and therefore don't need to return the results           
                     if self._should_push_pair(next_token):
                         self._push(TokenType.PAIR, f'{map_key.value}:{obj}')
                 else:
@@ -75,7 +77,9 @@ def get_result(self) -> Optional[bytes]:
     async def parse_token(self, token: str, value: str) -> None:
         token_type = TokenType.from_str(token)
         if token_type in VALUE_TOKENS:
-            self._handle_value_token(token_type, value)
+            val = self._handle_value_token(token_type, value)
+            if val is not None:
+                await self._handle_obj_emit(val)
         elif token_type == TokenType.MAP_KEY:
             self._handle_map_key_token(value)
         elif token_type in START_EVENTS:

…s/common/core/async_json_token_parser.py …rotocol/_core/async_json_token_parser.pycouchbase_analytics/common/core/async_json_token_parser.py renamed to acouchbase_analytics/protocol/_core/async_json_token_parser.py couchbase_analytics/common/core/async_json_token_parser.py renamed to acouchbase_analytics/protocol/_core/async_json_token_parser.py

@@ -16,19 +16,18 @@
 from __future__ import annotations
 
 import socket
-
-from typing import Optional, TYPE_CHECKING
+from typing import TYPE_CHECKING, Optional
 from uuid import uuid4
 
-from httpx import BasicAuth, AsyncClient, Response
+from httpx import URL, AsyncClient, BasicAuth, Response
 
 from couchbase_analytics.common.credential import Credential
 from couchbase_analytics.common.deserializer import Deserializer
 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
+    from couchbase_analytics.protocol._core.request import QueryRequest
 
 
 class _AsyncClientAdapter:
@@ -139,17 +138,22 @@ async def send_request(self, request: QueryRequest) -> Response:
         if not hasattr(self, '_client'):
             raise RuntimeError('Client not created yet')
 
-        if request.url is None:
-            raise ValueError('Request URL cannot be None')
+        # if request.url is None:
+        #     raise ValueError('Request URL cannot be None')
 
+        url = URL(scheme=request.url.scheme,
+                  host=request.url.host,
+                  port=request.url.port,
+                  path=request.url.path,)
         req = self._client.build_request(request.method,
-                                         request.url,
+                                         url,
                                          json=request.body,
                                          extensions=request.extensions)
         try:
             return await self._client.send(req, stream=True)
         except socket.gaierror as err:
-            raise RuntimeError(f'Unable to connect to {self._conn_details.get_scheme_host_and_port()}') from err
+            req_url = self._conn_details.url.get_formatted_url()
+            raise RuntimeError(f'Unable to connect to {req_url}') from err
 
     def reset_client(self) -> None:
         """

…nalytics/protocol/core/client_adapter.py …alytics/protocol/_core/client_adapter.pyacouchbase_analytics/protocol/core/client_adapter.py renamed to acouchbase_analytics/protocol/_core/client_adapter.py acouchbase_analytics/protocol/core/client_adapter.py renamed to acouchbase_analytics/protocol/_core/client_adapter.py

@@ -0,0 +1,57 @@
+#  Copyright 2016-2024. Couchbase, Inc.
+#  All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from __future__ import annotations
+
+import socket
+from ipaddress import IPv4Address, IPv6Address, ip_address
+from random import choice
+from typing import Optional, Set, Union
+
+import anyio
+
+from acouchbase_analytics.protocol.errors import ErrorMapper
+
+
+@ErrorMapper.handle_socket_error_async
+async def get_request_ip_async(host: str,
+                               port: int,
+                               previous_ips: Optional[Set[str]]=None) -> Optional[str]:
+    # Lets not call getaddrinfo, if the host is already an IP address
+    try:
+        ip: Optional[Union[IPv4Address, IPv6Address, str]] = ip_address(host)
+    except ValueError:
+        ip = None
+
+    # if we have localhost, httpx does not seem to be able to resolve IPv6 localhost (::1) properly
+    # TODO: IPv6 support for localhost??
+    if host == 'localhost':
+        ip = '127.0.0.1'
+
+    if previous_ips is None:
+        previous_ips = set()
+
+    if not ip:
+        result = await anyio.getaddrinfo(host, port, type=socket.SOCK_STREAM, family=socket.AF_UNSPEC)
+        try:
+            res_ip = choice([addr[4][0] for addr in result if addr[4][0] not in previous_ips])
+            ip = str(res_ip)
+        except IndexError:
+            ip = None
+    else:
+        ip_str = str(ip) if not isinstance(ip, str) else ip
+        ip = None if ip_str in previous_ips else ip_str
+
+    return ip