made merge_rows into an instance method

Author: daniel-sanche

google/cloud/bigtable/data/_async/_read_rows.py

@@ -15,7 +15,7 @@
 
 from __future__ import annotations
 
-from typing import Callable, Sequence, TYPE_CHECKING
+from typing import Sequence, TYPE_CHECKING
 
 import time
 
@@ -145,20 +145,20 @@ def _read_rows_attempt(self) -> CrossSync.Iterable[Row]:
                 )
             except _RowSetComplete:
                 # if we've already seen all the rows, we're done
-                return self.merge_rows(None, self._operation_metric, self._predicate)
+                return self.merge_rows(None)
         # revise the limit based on number of rows already yielded
         if self._remaining_count is not None:
             self.request.rows_limit = self._remaining_count
             if self._remaining_count == 0:
-                return self.merge_rows(None, self._operation_metric, self._predicate)
+                return self.merge_rows(None)
         # create and return a new row merger
         gapic_stream = self.target.client._gapic_client.read_rows(
             self.request,
             timeout=next(self.attempt_timeout_gen),
             retry=None,
         )
         chunked_stream = self.chunk_stream(gapic_stream)
-        return self.merge_rows(chunked_stream, self._operation_metric, self._predicate)
+        return self.merge_rows(chunked_stream)
 
     @CrossSync.convert()
     async def chunk_stream(
@@ -217,9 +217,7 @@ async def chunk_stream(
         replace_symbols={"__aiter__": "__iter__", "__anext__": "__next__"},
     )
     async def merge_rows(
-        chunks: CrossSync.Iterable[ReadRowsResponsePB.CellChunk] | None,
-        operation_metric: ActiveOperationMetric,
-        retryable_predicate: Callable[[Exception], bool],
+        self, chunks: CrossSync.Iterable[ReadRowsResponsePB.CellChunk] | None
     ) -> CrossSync.Iterable[Row]:
         """
         Merge chunks into rows
@@ -231,7 +229,7 @@ async def merge_rows(
         """
         try:
             if chunks is None:
-                operation_metric.end_with_success()
+                self._operation_metric.end_with_success()
                 return
             it = chunks.__aiter__()
             # For each row
@@ -240,7 +238,7 @@ async def merge_rows(
                     c = await it.__anext__()
                 except CrossSync.StopIteration:
                     # stream complete
-                    operation_metric.end_with_success()
+                    self._operation_metric.end_with_success()
                     return
                 row_key = c.row_key
 
@@ -321,8 +319,8 @@ async def merge_rows(
                             yield Row(row_key, cells)
                             # most metric operations use setters, but this one updates
                             # the value directly to avoid extra overhead
-                            if operation_metric.active_attempt is not None:
-                                operation_metric.active_attempt.application_blocking_time_ns += (  # type: ignore
+                            if self._operation_metric.active_attempt is not None:
+                                self._operation_metric.active_attempt.application_blocking_time_ns += (  # type: ignore
                                     time.monotonic_ns() - block_time
                                 ) * 1000
                             break
@@ -342,11 +340,11 @@ async def merge_rows(
                 except CrossSync.StopIteration:
                     raise InvalidChunk("premature end of stream")
         except Exception as generic_exception:
-            if not retryable_predicate(generic_exception):
-                operation_metric.end_attempt_with_status(generic_exception)
+            if not self._predicate(generic_exception):
+                self._operation_metric.end_attempt_with_status(generic_exception)
             raise generic_exception
         else:
-            operation_metric.end_with_success()
+            self._operation_metric.end_with_success()
 
     @staticmethod
     def _revise_request_rowset(

google/cloud/bigtable/data/_async/client.py

@@ -87,7 +87,6 @@
 from google.cloud.bigtable.data.row_filters import RowFilterChain
 from google.cloud.bigtable.data._metrics import BigtableClientSideMetricsController
 from google.cloud.bigtable.data._metrics import OperationType
-from google.cloud.bigtable.data._metrics.handlers._stdout import _StdoutMetricsHandler
 
 from google.cloud.bigtable.data._cross_sync import CrossSync
 
@@ -944,7 +943,7 @@ def __init__(
 
         self._metrics = BigtableClientSideMetricsController(
             client._metrics_interceptor,
-            handlers=[_StdoutMetricsHandler()],
+            handlers=[],
             project_id=self.client.project,
             instance_id=instance_id,
             table_id=table_id,

google/cloud/bigtable/data/_metrics/data_model.py

@@ -396,7 +396,6 @@ def _handle_error(message: str) -> None:
         """
         full_message = f"Error in Bigtable Metrics: {message}"
         LOGGER.warning(full_message)
-        raise RuntimeError(full_message)
 
     def __enter__(self):
         """

google/cloud/bigtable/data/_sync_autogen/_read_rows.py

@@ -17,7 +17,7 @@
 # This file is automatically generated by CrossSync. Do not edit manually.
 
 from __future__ import annotations
-from typing import Callable, Sequence, TYPE_CHECKING
+from typing import Sequence, TYPE_CHECKING
 import time
 from google.cloud.bigtable_v2.types import ReadRowsRequest as ReadRowsRequestPB
 from google.cloud.bigtable_v2.types import ReadRowsResponse as ReadRowsResponsePB
@@ -126,16 +126,16 @@ def _read_rows_attempt(self) -> CrossSync._Sync_Impl.Iterable[Row]:
                     last_seen_row_key=self._last_yielded_row_key,
                 )
             except _RowSetComplete:
-                return self.merge_rows(None, self._operation_metric, self._predicate)
+                return self.merge_rows(None)
         if self._remaining_count is not None:
             self.request.rows_limit = self._remaining_count
             if self._remaining_count == 0:
-                return self.merge_rows(None, self._operation_metric, self._predicate)
+                return self.merge_rows(None)
         gapic_stream = self.target.client._gapic_client.read_rows(
             self.request, timeout=next(self.attempt_timeout_gen), retry=None
         )
         chunked_stream = self.chunk_stream(gapic_stream)
-        return self.merge_rows(chunked_stream, self._operation_metric, self._predicate)
+        return self.merge_rows(chunked_stream)
 
     def chunk_stream(
         self,
@@ -182,9 +182,7 @@ def chunk_stream(
 
     @staticmethod
     def merge_rows(
-        chunks: CrossSync._Sync_Impl.Iterable[ReadRowsResponsePB.CellChunk] | None,
-        operation_metric: ActiveOperationMetric,
-        retryable_predicate: Callable[[Exception], bool],
+        self, chunks: CrossSync._Sync_Impl.Iterable[ReadRowsResponsePB.CellChunk] | None
     ) -> CrossSync._Sync_Impl.Iterable[Row]:
         """Merge chunks into rows
 
@@ -194,14 +192,14 @@ def merge_rows(
             Row: the next row in the stream"""
         try:
             if chunks is None:
-                operation_metric.end_with_success()
+                self._operation_metric.end_with_success()
                 return
             it = chunks.__iter__()
             while True:
                 try:
                     c = it.__next__()
                 except CrossSync._Sync_Impl.StopIteration:
-                    operation_metric.end_with_success()
+                    self._operation_metric.end_with_success()
                     return
                 row_key = c.row_key
                 if not row_key:
@@ -268,8 +266,8 @@ def merge_rows(
                         if c.commit_row:
                             block_time = time.monotonic_ns()
                             yield Row(row_key, cells)
-                            if operation_metric.active_attempt is not None:
-                                operation_metric.active_attempt.application_blocking_time_ns += (
+                            if self._operation_metric.active_attempt is not None:
+                                self._operation_metric.active_attempt.application_blocking_time_ns += (
                                     time.monotonic_ns() - block_time
                                 ) * 1000
                             break
@@ -289,11 +287,11 @@ def merge_rows(
                 except CrossSync._Sync_Impl.StopIteration:
                     raise InvalidChunk("premature end of stream")
         except Exception as generic_exception:
-            if not retryable_predicate(generic_exception):
-                operation_metric.end_attempt_with_status(generic_exception)
+            if not self._predicate(generic_exception):
+                self._operation_metric.end_attempt_with_status(generic_exception)
             raise generic_exception
         else:
-            operation_metric.end_with_success()
+            self._operation_metric.end_with_success()
 
     @staticmethod
     def _revise_request_rowset(row_set: RowSetPB, last_seen_row_key: bytes) -> RowSetPB:

google/cloud/bigtable/data/_sync_autogen/client.py

@@ -74,7 +74,6 @@
 from google.cloud.bigtable.data.row_filters import RowFilterChain
 from google.cloud.bigtable.data._metrics import BigtableClientSideMetricsController
 from google.cloud.bigtable.data._metrics import OperationType
-from google.cloud.bigtable.data._metrics.handlers._stdout import _StdoutMetricsHandler
 from google.cloud.bigtable.data._cross_sync import CrossSync
 from typing import Iterable
 from grpc import insecure_channel
@@ -735,7 +734,7 @@ def __init__(
         )
         self._metrics = BigtableClientSideMetricsController(
             client._metrics_interceptor,
-            handlers=[_StdoutMetricsHandler()],
+            handlers=[],
             project_id=self.client.project,
             instance_id=instance_id,
             table_id=table_id,