fixed lint

Author: daniel-sanche

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

@@ -106,10 +106,8 @@ def __init__(
             self.is_retryable,
             metric.backoff_generator,
             operation_timeout,
-            exception_factory=self._operation_metric.track_terminal_error(
-                _retry_exception_factory
-            ),
-            on_error=self._operation_metric.track_retryable_error,
+            exception_factory=metric.track_terminal_error(_retry_exception_factory),
+            on_error=metric.track_retryable_error,
         )
         # initialize state
         self.timeout_generator = _attempt_timeout_generator(

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

@@ -350,8 +350,6 @@ async def merge_rows(
         except Exception as generic_exception:
             # handle exceptions in retry wrapper
             raise generic_exception
-        else:
-            self._operation_metric.end_with_success()
 
     @staticmethod
     def _revise_request_rowset(

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

@@ -67,6 +67,7 @@ def wrapper(self, continuation, client_call_details, request):
 async def _get_metadata(source) -> dict[str, str | bytes] | None:
     """Helper to extract metadata from a call or RpcError"""
     try:
+        metadata: Sequence[tuple[str, str | bytes]]
         if CrossSync.is_async:
             # grpc.aio returns metadata in Metadata objects
             if isinstance(source, AioRpcError):
@@ -79,11 +80,9 @@ async def _get_metadata(source) -> dict[str, str | bytes] | None:
                 )
         else:
             # sync grpc returns metadata as a sequence of tuples
-            metadata: Sequence[tuple[str.str | bytes]] = (
-                source.trailing_metadata() + source.initial_metadata()
-            )
+            metadata = source.trailing_metadata() + source.initial_metadata()
         # convert metadata to dict format
-        return {k: v for k, v in metadata}
+        return {k: v for (k, v) in metadata}
     except Exception:
         # ignore errors while fetching metadata
         return None

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

@@ -13,7 +13,7 @@
 # limitations under the License.
 from __future__ import annotations
 
-from typing import ClassVar, Tuple, cast, TYPE_CHECKING
+from typing import Callable, ClassVar, List, Tuple, Optional, cast, TYPE_CHECKING
 
 import time
 import re
@@ -54,6 +54,11 @@
 
 INVALID_STATE_ERROR = "Invalid state for {}: {}"
 
+ExceptionFactoryType = Callable[
+    [List[Exception], RetryFailureReason, Optional[float]],
+    Tuple[Exception, Optional[Exception]],
+]
+
 
 class OperationType(Enum):
     """Enum for the type of operation being performed."""
@@ -168,7 +173,7 @@ class ActiveOperationMetric:
     flow_throttling_time_ns: int = 0
 
     _active_operation_context: ClassVar[
-        contextvars.ContextVar
+        contextvars.ContextVar[ActiveOperationMetric]
     ] = contextvars.ContextVar("active_operation_context")
 
     @classmethod
@@ -419,12 +424,8 @@ def track_retryable_error(self, exc: Exception) -> None:
             self.end_attempt_with_status(exc)
 
     def track_terminal_error(
-        self,
-        exception_factory: callable[
-            [list[Exception], RetryFailureReason, float | None],
-            tuple[Exception, Exception | None],
-        ],
-    ) -> callable[[list[Exception], RetryFailureReason, float | None], None]:
+        self, exception_factory: ExceptionFactoryType
+    ) -> ExceptionFactoryType:
         """
         Used as input to api_core.Retry classes, to track when terminal errors are encountered
 

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

@@ -84,10 +84,8 @@ def __init__(
             self.is_retryable,
             metric.backoff_generator,
             operation_timeout,
-            exception_factory=self._operation_metric.track_terminal_error(
-                _retry_exception_factory
-            ),
-            on_error=self._operation_metric.track_retryable_error,
+            exception_factory=metric.track_terminal_error(_retry_exception_factory),
+            on_error=metric.track_retryable_error,
         )
         self.timeout_generator = _attempt_timeout_generator(
             attempt_timeout, operation_timeout

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

@@ -294,8 +294,6 @@ def merge_rows(
             raise close_exception
         except Exception as generic_exception:
             raise generic_exception
-        else:
-            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/metrics_interceptor.py

@@ -49,9 +49,8 @@ def wrapper(self, continuation, client_call_details, request):
 def _get_metadata(source) -> dict[str, str | bytes] | None:
     """Helper to extract metadata from a call or RpcError"""
     try:
-        metadata: Sequence[tuple[str.str | bytes]] = (
-            source.trailing_metadata() + source.initial_metadata()
-        )
+        metadata: Sequence[tuple[str, str | bytes]]
+        metadata = source.trailing_metadata() + source.initial_metadata()
         return {k: v for (k, v) in metadata}
     except Exception:
         return None