mypy

Author: gkevinzheng

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

@@ -14,7 +14,7 @@
 #
 from __future__ import annotations
 
-from typing import Sequence, TYPE_CHECKING, cast
+from typing import Callable, Optional, Sequence, TYPE_CHECKING, cast
 import atexit
 import warnings
 from collections import deque
@@ -276,7 +276,9 @@ def __init__(
         self._newest_exceptions: deque[Exception] = deque(
             maxlen=self._exception_list_limit
         )
-        self._user_batch_completed_callback = None
+        self._user_batch_completed_callback: Optional[
+            Callable[[list[status_pb2.Status]], None]
+        ] = None
         # clean up on program exit
         atexit.register(self._on_exit)
 

google/cloud/bigtable/data/_helpers.py

@@ -16,7 +16,7 @@
 """
 from __future__ import annotations
 
-from typing import Callable, Sequence, List, Optional, Tuple, TYPE_CHECKING, Union
+from typing import cast, Callable, Sequence, List, Optional, Tuple, TYPE_CHECKING, Union
 import time
 import enum
 from collections import namedtuple
@@ -272,14 +272,17 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status:
     Returns:
         status_pb2.Status: A Status proto object.
     """
-    if (
-        isinstance(exc, core_exceptions.GoogleAPICallError)
-        and exc.grpc_status_code is not None
-    ):
-        return status_pb2.Status(  # type: ignore[unreachable]
-            code=exc.grpc_status_code.value[0],
-            message=exc.message,
-            details=exc.details,
+    if isinstance(exc, core_exceptions.GoogleAPICallError):
+        status_code = cast(Optional["grpc.StatusCode"], exc.grpc_status_code)
+        if status_code is not None:
+            return status_pb2.Status(
+                code=status_code.value[0],
+                message=exc.message,
+                details=exc.details,
+            )
+        return status_pb2.Status(
+            code=code_pb2.Code.UNKNOWN,
+            message="An unknown error has occurred",
         )
 
     return status_pb2.Status(

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

@@ -16,7 +16,7 @@
 # This file is automatically generated by CrossSync. Do not edit manually.
 
 from __future__ import annotations
-from typing import Sequence, TYPE_CHECKING, cast
+from typing import Callable, Optional, Sequence, TYPE_CHECKING, cast
 import atexit
 import warnings
 from collections import deque
@@ -238,7 +238,9 @@ def __init__(
         self._newest_exceptions: deque[Exception] = deque(
             maxlen=self._exception_list_limit
         )
-        self._user_batch_completed_callback = None
+        self._user_batch_completed_callback: Optional[
+            Callable[[list[status_pb2.Status]], None]
+        ] = None
         atexit.register(self._on_exit)
 
     def _timer_routine(self, interval: float | None) -> None: