Updates to move SDK closer to initial developer preview.

Changes
=======
* Add retry logic
* Cleaned up error handling
* Reorganized RequestContext logic
* Updated tests
* Updated pre-commit configuration and ensured all stages are passing

Author: thejcfactor

.pre-commit-config.yaml

@@ -20,6 +20,54 @@ repos:
       # - id: ruff-format
       #   types_or: [ python, pyi ]
       # Compile requirements
+  - repo: https://github.com/PyCQA/bandit
+    rev: 1.8.6
+    hooks:
+      - id: bandit
+        exclude: |
+          (?x)^(
+              acouchbase_analytics/tests/|
+              couchbase_analytics/tests/|
+              tests/|
+              couchbase_analytics_version.py
+          )
+        args:
+          [
+              --quiet
+          ]
+  - repo: https://github.com/PyCQA/isort
+    rev: 5.13.2
+    hooks:
+      - id: isort
+        exclude: |
+          (?x)^(
+              deps/|
+              src/
+          )
+        args:
+          [
+            "--multi-line 1",
+            "--force-grid-wrap 3",
+            "--use-parentheses True",
+            "--ensure-newline-before-comments True",
+            "--line-length 120",
+            "--order-by-type True"
+          ]
+  - repo: local
+    hooks:
+      - id: mypy
+        name: mypy
+        entry: "./run-mypy"
+        language: python
+        additional_dependencies:
+          - mypy~=1.16.1
+          - pytest~=8.3.5
+          - httpx~=0.28.1
+          - aiohttp~=3.11.10
+        types:
+          - python
+        require_serial: true
+        verbose: true
   - repo: https://github.com/astral-sh/uv-pre-commit
     # uv version.
     rev: 0.7.19

acouchbase_analytics/cluster.py

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

acouchbase_analytics/cluster.pyi

@@ -23,7 +23,10 @@ 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:

acouchbase_analytics/database.py

@@ -23,7 +23,6 @@
 else:
     from typing import TypeAlias
 
-
 from acouchbase_analytics.scope import AsyncScope
 
 if TYPE_CHECKING:

acouchbase_analytics/protocol/_core/async_json_stream.py

@@ -18,15 +18,14 @@
 from typing import AsyncIterator, Optional
 
 import ijson
-from anyio import EndOfStream, Event, create_memory_object_stream
+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._core.json_parsing import (JsonStreamConfig,
+                                                           ParsedResult,
+                                                           ParsedResultType)
 from couchbase_analytics.common.errors import AnalyticsError
 
 
@@ -133,7 +132,11 @@ async def _process_token_stream(self) -> None:
             except StopAsyncIteration:
                 self._token_stream_exhausted = True
             except ijson.common.IncompleteJSONError as ex:
-                raise JsonParsingError(cause=ex) from None
+                # TODO: log this error
+                self._token_stream_exhausted = True
+                await self._send_to_stream(ParsedResult(str(ex).encode('utf-8'), ParsedResultType.ERROR), close=True)
+                self._handle_notification(ParsedResultType.ERROR)
+                return
 
 
         if self._token_stream_exhausted:

acouchbase_analytics/protocol/_core/async_json_token_parser.py

@@ -15,16 +15,18 @@
 
 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 (POP_EVENTS,
+                                                                     START_EVENTS,
+                                                                     VALUE_TOKENS,
+                                                                     JsonTokenParserBase,
+                                                                     ParsingState,
+                                                                     TokenType)
 
 
 class AsyncJsonTokenParser(JsonTokenParserBase):

acouchbase_analytics/protocol/_core/client_adapter.py

@@ -19,7 +19,10 @@
 from typing import TYPE_CHECKING, Optional
 from uuid import uuid4
 
-from httpx import URL, AsyncClient, BasicAuth, Response
+from httpx import (URL,
+                   AsyncClient,
+                   BasicAuth,
+                   Response)
 
 from couchbase_analytics.common.credential import Credential
 from couchbase_analytics.common.deserializer import Deserializer
@@ -138,11 +141,8 @@ 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')
-
         url = URL(scheme=request.url.scheme,
-                  host=request.url.host,
+                  host=request.url.ip,
                   port=request.url.port,
                   path=request.url.path,)
         req = self._client.build_request(request.method,

acouchbase_analytics/protocol/_core/net_utils.py

@@ -16,19 +16,19 @@
 from __future__ import annotations
 
 import socket
-from ipaddress import IPv4Address, IPv6Address, ip_address
+from ipaddress import (IPv4Address,
+                       IPv6Address,
+                       ip_address)
 from random import choice
-from typing import Optional, Set, Union
+from typing import Optional, 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]:
+async def get_request_ip_async(host: str, port: int) -> str:
     # Lets not call getaddrinfo, if the host is already an IP address
     try:
         ip: Optional[Union[IPv4Address, IPv6Address, str]] = ip_address(host)
@@ -40,18 +40,11 @@ async def get_request_ip_async(host: str,
     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
+        res_ip = choice([addr[4][0] for addr in result])  # nosec B311
+        ip = str(res_ip)
     else:
-        ip_str = str(ip) if not isinstance(ip, str) else ip
-        ip = None if ip_str in previous_ips else ip_str
+        ip = str(ip)
 
     return ip

acouchbase_analytics/protocol/_core/request_context.py

@@ -3,19 +3,33 @@
 import json
 from asyncio import CancelledError, Task
 from types import TracebackType
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, Coroutine, Dict, List, Optional, Type, Union
+from typing import (TYPE_CHECKING,
+                    Any,
+                    Awaitable,
+                    Callable,
+                    Coroutine,
+                    Dict,
+                    List,
+                    Optional,
+                    Type,
+                    Union)
 from uuid import uuid4
 
 import anyio
 from httpx import Response as HttpCoreResponse
 from httpx import TimeoutException
 
-from acouchbase_analytics.protocol._core.anyio_utils import AsyncBackend, current_async_library, get_time
+from acouchbase_analytics.protocol._core.anyio_utils import (AsyncBackend,
+                                                             current_async_library,
+                                                             get_time)
 from acouchbase_analytics.protocol._core.async_json_stream import AsyncJsonStream
 from acouchbase_analytics.protocol._core.net_utils import get_request_ip_async
-from couchbase_analytics.common._core import JsonStreamConfig, ParsedResult, ParsedResultType
+from couchbase_analytics.common._core import (JsonStreamConfig,
+                                              ParsedResult,
+                                              ParsedResultType)
 from couchbase_analytics.common._core.error_context import ErrorContext
-from couchbase_analytics.common.errors import AnalyticsError, InvalidCredentialError
+from couchbase_analytics.common.backoff_calculator import DefaultBackoffCalculator
+from couchbase_analytics.common.errors import AnalyticsError
 from couchbase_analytics.common.streaming import StreamingState
 from couchbase_analytics.protocol.connection import DEFAULT_TIMEOUTS
 from couchbase_analytics.protocol.errors import ErrorMapper
@@ -37,6 +51,7 @@ def __init__(self,
         self._client_adapter = client_adapter
         self._request = request
         self._backend = backend or current_async_library()
+        self._backoff_calc = DefaultBackoffCalculator()
         self._error_ctx = ErrorContext(num_attempts=0,
                                        method=request.method,
                                        statement=request.get_request_statement())
@@ -85,6 +100,10 @@ def request_error(self) -> Optional[Union[BaseException, Exception]]:
     def request_state(self) -> StreamingState:
         return self._request_state
 
+    @property
+    def retry_limit_exceeded(self) -> bool:
+        return self.error_context.num_attempts > self._request.max_retries
+
     @property
     def results_or_errors_type(self) -> ParsedResultType:
         return self._json_stream.results_or_errors_type
@@ -137,10 +156,12 @@ def _maybe_set_request_error(self,
             self._request_error = exc_val
 
     async def _process_error(self,
-                            json_data: List[Dict[str, Any]],
+                            json_data: Union[str, List[Dict[str, Any]]],
                             handle_context_shutdown: Optional[bool]=False) -> None:
         self._request_state = StreamingState.Error
-        if not isinstance(json_data, list):
+        if isinstance(json_data, str):
+            self._request_error = ErrorMapper.build_error_from_http_status_code(json_data, self._error_ctx)
+        elif not isinstance(json_data, list):
             self._request_error = AnalyticsError('Cannot parse error response; expected JSON array',
                                                  context=str(self._error_ctx))
         else:
@@ -154,7 +175,6 @@ def _reset_stream(self) -> None:
         if hasattr(self, '_json_stream'):
             del self._json_stream
         self._request_state = StreamingState.ResetAndNotStarted
-        self._request.previous_ips = set()
         self._stage_completed = None
         self._cancel_scope_deadline_updated = False
 
@@ -197,6 +217,9 @@ async def _wait_for_stage_to_complete(self) -> None:
             return
         await self._stage_completed.wait()
 
+    def calculate_backoff(self) -> float:
+        return self._backoff_calc.calculate_backoff(self._error_ctx.num_attempts) / 1000
+
     def cancel_request(self,
                        fn: Optional[Callable[..., Awaitable[Any]]]=None,
                        *args: object) -> None:
@@ -273,6 +296,9 @@ def okay_to_delay_and_retry(self, delay: float) -> bool:
         if will_time_out:
             self._request_state = StreamingState.Timeout
             return False
+        elif self.retry_limit_exceeded:
+            self._request_state = StreamingState.Error
+            return False
         else:
             self._reset_stream()
             return True
@@ -296,10 +322,15 @@ async def process_response(self,
         # we have all the data, close the core response/stream
         await close_handler()
 
-        json_response = json.loads(raw_response.value)
-        if 'errors' in json_response:
-            await self._process_error(json_response['errors'], handle_context_shutdown=handle_context_shutdown)
-        return json_response
+        try:
+            json_response = json.loads(raw_response.value)
+        except json.JSONDecodeError:
+            await self._process_error(str(raw_response.value),
+                                      handle_context_shutdown=handle_context_shutdown)
+        else:
+            if 'errors' in json_response:
+                await self._process_error(json_response['errors'], handle_context_shutdown=handle_context_shutdown)
+            return json_response
 
     async def reraise_after_shutdown(self, err: Exception) -> None:
         try:
@@ -309,24 +340,17 @@ async def reraise_after_shutdown(self, err: Exception) -> None:
             raise ex from None
 
     async def send_request(self, enable_trace_handling: Optional[bool]=False) -> HttpCoreResponse:
-        ip = await get_request_ip_async(self._request.url.host, self._request.url.port, self._request.previous_ips)
-        if ip is None:
-            attempted_ips = ', '.join(self._request.previous_ips or [])
-            raise AnalyticsError(message=f'Connect failure.  Unable to connect to any resolved IPs: {attempted_ips}.',
-                                 context=str(self._error_ctx))
-
+        self._error_ctx.update_num_attempts()
+        ip = await get_request_ip_async(self._request.url.host, self._request.url.port)
         if enable_trace_handling is True:
             (self._request.update_url(ip, self._client_adapter.analytics_path)
-                          .add_trace_to_extensions(self._trace_handler)
-                          .update_previous_ips(ip))
+                          .add_trace_to_extensions(self._trace_handler))
         else:
-            self._request.update_url(ip, self._client_adapter.analytics_path).update_previous_ips(ip)
+            self._request.update_url(ip, self._client_adapter.analytics_path)
         # TODO:  add logging; provide request details (to/from, deadlines, etc.)
         self._error_ctx.update_request_context(self._request)
         response = await self._client_adapter.send_request(self._request)
         self._error_ctx.update_response_context(response)
-        if response.status_code == 401:
-            raise InvalidCredentialError(context=str(self._error_ctx))
         return response
 
     async def shutdown(self,