PYCBC-1756: Streaming API Improvements

Motivation
==========
Some final cleanup/improvement changes to the streaming API prior to
releasing the next dot-minor that includes significant changes to the
underlying c-extension and the overall shape of the SDK's internal
logic.

Modification
============
* Handle metadata metrics that might be set to `None`
* Use default C++ core timeouts for streaming timeouts (if not provided)
* Update query params property to not JSON-ify positional, named and raw
  options each time it is called
* Add query test to confirm metrics handling is correct

Change-Id: I42bf644a1085fbc812fe0a3f94437b96c17bf2be
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/242421
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Brett Lawson <brett19@gmail.com>

Author: thejcfactor

acouchbase/logic/cluster_impl.py

@@ -57,6 +57,7 @@ class AsyncClusterImpl:
     def __init__(self, connstr: str, *options: object, **kwargs: object) -> None:
         loop: Optional[AbstractEventLoop] = kwargs.pop('loop', None)
         loop_validator = kwargs.pop('loop_validator', None)
+        kwargs['_default_timeouts'] = pycbc_connection.pycbc_get_default_timeouts()
         self._cluster_settings = ClusterSettings.build_cluster_settings(connstr, *options, **kwargs)
         connect_request = CreateConnectionRequest(self._cluster_settings.connstr,
                                                   self._cluster_settings.auth,

couchbase/logic/analytics.py

@@ -151,8 +151,9 @@ def errors(self) -> List[AnalyticsError]:
         )
 
     def metrics(self) -> Optional[AnalyticsMetrics]:
-        if "metrics" in self._raw:
-            return AnalyticsMetrics(self._raw.get("metrics", {}))
+        raw_metrics = self._raw.get('metrics', None)
+        if raw_metrics:
+            return AnalyticsMetrics(raw_metrics)
         return None
 
     def __repr__(self):
@@ -166,6 +167,7 @@ class AnalyticsQuery:
         'read_only': {'readonly': lambda x: x},
         'scan_consistency': {'consistency': lambda x: x.value},
         'client_context_id': {'client_context_id': lambda x: x},
+        'metrics': {'metrics': lambda x: x},
         'priority': {'priority': lambda x: x},
         'query_context': {'query_context': lambda x: x},
         'serializer': {'serializer': lambda x: x},
@@ -196,10 +198,10 @@ def _set_named_args(self, **kv):
             `$` identifier.
 
         """
+        arg_dict = self._params.setdefault("named_parameters", {})
         # C++ core wants all args JSONified bytes
         named_params = {f'${k}': json.dumps(v).encode('utf-8') for k, v in kv.items()}
-
-        self._params["named_parameters"] = named_params
+        arg_dict.update(named_params)
         return self
 
     def _add_pos_args(self, *args):

couchbase/logic/cluster_impl.py

@@ -57,6 +57,7 @@
 class ClusterImpl:
     def __init__(self, connstr: str, *options: object, **kwargs: object) -> None:
         skip_connect = kwargs.pop('skip_connect', None)
+        kwargs['_default_timeouts'] = pycbc_connection.pycbc_get_default_timeouts()
         self._cluster_settings = ClusterSettings.build_cluster_settings(connstr, *options, **kwargs)
         connect_request = CreateConnectionRequest(self._cluster_settings.connstr,
                                                   self._cluster_settings.auth,

couchbase/logic/cluster_settings.py

@@ -335,6 +335,7 @@ class ClusterSettings:
     cluster_options: Dict[str, Any]
     default_transcoder: Transcoder
     default_serializer: Serializer
+    default_timeouts: Dict[str, int]
     metrics_options: Dict[str, Any]
     orphan_options: Dict[str, Any]
     streaming_timeouts: StreamingTimeouts
@@ -352,6 +353,7 @@ def build_cluster_settings(cls,
                                *options,  # type: ClusterOptions
                                **kwargs  # type: Dict[str, Any]
                                ) -> ClusterSettings:
+        default_timeouts = kwargs.pop('_default_timeouts', {})
         # parse query string prior to parsing ClusterOptions
         connection_str, query_opts, legacy_opts = parse_connection_string(connstr)
 
@@ -375,10 +377,11 @@ def build_cluster_settings(cls,
 
         timeout_opts = build_timeout_options(cluster_opts)
         streaming_timeouts: StreamingTimeouts = {
-            'analytics_timeout': timeout_opts.get('analytics_timeout', None),
-            'query_timeout': timeout_opts.get('query_timeout', None),
-            'search_timeout': timeout_opts.get('search_timeout', None),
-            'view_timeout': timeout_opts.get('view_timeout', None),
+            'analytics_timeout': timeout_opts.get('analytics_timeout',
+                                                  default_timeouts.get('analytics_timeout', None)),
+            'query_timeout': timeout_opts.get('query_timeout', default_timeouts.get('query_timeout', None)),
+            'search_timeout': timeout_opts.get('search_timeout', default_timeouts.get('search_timeout', None)),
+            'view_timeout': timeout_opts.get('view_timeout', default_timeouts.get('view_timeout', None)),
         }
         tracing_opts, orphan_opts, tracer = build_tracing_and_orphan_options(cluster_opts)
         metrics_opts, meter = build_metrics_options(cluster_opts)
@@ -389,6 +392,7 @@ def build_cluster_settings(cls,
                    cluster_opts,
                    default_transcoder,
                    default_serializer,
+                   default_timeouts,
                    metrics_opts,
                    orphan_opts,
                    streaming_timeouts,

couchbase/logic/n1ql.py

@@ -15,7 +15,6 @@
 
 from __future__ import annotations
 
-import functools
 import json
 from datetime import timedelta
 from enum import Enum
@@ -363,7 +362,10 @@ def _set_named_args(self, **kv):
 
         """
         arg_dict = self._params.setdefault("named_parameters", {})
-        arg_dict.update(kv)
+        # C++ core wants all args JSONified bytes
+        named_params = {f'${k}': json.dumps(v).encode('utf-8') for k, v in kv.items()}
+        arg_dict.update(named_params)
+        return self
 
     def _add_pos_args(self, *args):
         """
@@ -372,7 +374,9 @@ def _add_pos_args(self, *args):
         :param args: Values to be used
         """
         arg_array = self._params.setdefault("positional_parameters", [])
-        arg_array.extend(args)
+        # C++ core wants all args JSONified bytes
+        json_args = [json.dumps(arg).encode('utf-8') for arg in args]
+        arg_array.extend(json_args)
 
     def set_option(self, name, value):
         """
@@ -386,24 +390,9 @@ def set_option(self, name, value):
         """
         self._params[name] = value
 
-    @functools.cached_property
-    def params(self) -> Dict[str, Any]:
-        params = self._params
-
-        # couchbase++ wants all args JSONified,
-        # For now encode to bytes to make couchbase::json_string <--> std::vector<std::byte> easier
-        raw = params.pop('raw', None)
-        if raw:
-            params['raw'] = {f'{k}': self._serializer.serialize(v) for k, v in raw.items()}
-
-        positional_args = params.pop('positional_parameters', None)
-        if positional_args:
-            params['positional_parameters'] = [self._serializer.serialize(arg) for arg in positional_args]
-
-        named_params = params.pop('named_parameters', None)
-        if named_params:
-            params['named_parameters'] = {f'${k}': self._serializer.serialize(v) for k, v in named_params.items()}
-        return params
+    @property
+    def params(self):
+        return self._params
 
     @property
     def metrics(self) -> bool:
@@ -668,7 +657,8 @@ def raw(self, value  # type: Dict[str, Any]
         for k in value.keys():
             if not isinstance(k, str):
                 raise TypeError("key for raw value must be str")
-        self.set_option('raw', value)
+        raw_params = {f'{k}': json.dumps(v).encode('utf-8') for k, v in value.items()}
+        self.set_option('raw', raw_params)
 
     @property
     def span(self) -> Optional[SpanProtocol]:

couchbase/logic/search.py

@@ -366,8 +366,9 @@ def errors(self) -> Dict[str, str]:
         return self._raw.get('errors', {})
 
     def metrics(self) -> Optional[SearchMetrics]:
-        if 'metrics' in self._raw:
-            return SearchMetrics(self._raw.get('metrics', {}))
+        raw_metrics = self._raw.get('metrics', None)
+        if raw_metrics:
+            return SearchMetrics(raw_metrics)
         return None
 
     def client_context_id(self) -> Optional[str]:

couchbase/tests/query_t.py

@@ -48,6 +48,7 @@ class QueryCollectionTestSuite:
         'test_query_fully_qualified',
         'test_query_in_thread',
         'test_query_metadata',
+        'test_query_metadata_no_metrics',
         'test_query_ryow',
         'test_query_with_metrics',
         'test_scope_query',
@@ -134,6 +135,28 @@ def test_query_metadata(self, cb_env):
             assert isinstance(warning.message(), str)
             assert isinstance(warning.code(), int)
 
+    def test_query_metadata_no_metrics(self, cb_env):
+        result = cb_env.scope.query(f"SELECT * FROM `{cb_env.collection.name}` LIMIT 2",
+                                    QueryOptions(metrics=False))
+        cb_env.assert_rows(result, 2)
+        metadata = result.metadata()  # type: QueryMetaData
+        for id_meth in (metadata.client_context_id, metadata.request_id):
+            id_res = id_meth()
+            fail_msg = "{} failed".format(id_meth)
+            assert isinstance(id_res, str), fail_msg
+
+        metrics = metadata.metrics()
+        assert metrics is None
+
+        assert metadata.status() == QueryStatus.SUCCESS
+        assert isinstance(metadata.signature(), (str, dict))
+        assert isinstance(metadata.warnings(), list)
+
+        for warning in metadata.warnings():
+            assert isinstance(warning, QueryWarning)
+            assert isinstance(warning.message(), str)
+            assert isinstance(warning.code(), int)
+
     def test_query_ryow(self, cb_env):
         key, value = cb_env.get_new_doc()
         result = cb_env.scope.query(f'SELECT * FROM `{cb_env.collection.name}` USE KEYS "{key}"')
@@ -199,6 +222,7 @@ class QueryTestSuite:
         'test_query_error_context',
         'test_query_in_thread',
         'test_query_metadata',
+        'test_query_metadata_no_metrics',
         'test_query_raw_options',
         'test_query_ryow',
         'test_query_timeout',
@@ -319,6 +343,28 @@ def test_query_metadata(self, cb_env):
             assert isinstance(warning.message(), str)
             assert isinstance(warning.code(), int)
 
+    def test_query_metadata_no_metrics(self, cb_env):
+        result = cb_env.cluster.query(f"SELECT * FROM `{cb_env.bucket.name}` LIMIT 2",
+                                      QueryOptions(metrics=False))
+        cb_env.assert_rows(result, 2)
+        metadata = result.metadata()  # type: QueryMetaData
+        for id_meth in (metadata.client_context_id, metadata.request_id):
+            id_res = id_meth()
+            fail_msg = "{} failed".format(id_meth)
+            assert isinstance(id_res, str), fail_msg
+
+        metrics = metadata.metrics()
+        assert metrics is None
+
+        assert metadata.status() == QueryStatus.SUCCESS
+        assert isinstance(metadata.signature(), (str, dict))
+        assert isinstance(metadata.warnings(), list)
+
+        for warning in metadata.warnings():
+            assert isinstance(warning, QueryWarning)
+            assert isinstance(warning.message(), str)
+            assert isinstance(warning.code(), int)
+
     def test_query_raw_options(self, cb_env):
         # via raw, we should be able to pass any option
         # if using named params, need to match full name param in query