Merge branch 'main' into renovate/all

Author: parthea

google/api_core/bidi.py

@@ -306,6 +306,8 @@ def close(self):
         self._request_queue.put(None)
         self.call.cancel()
         self._request_generator = None
+        self._initial_request = None
+        self._callbacks = []
         # Don't set self.call to None. Keep it around so that send/recv can
         # raise the error.
 
@@ -717,6 +719,7 @@ def stop(self):
                     _LOGGER.warning("Background thread did not exit.")
 
             self._thread = None
+            self._on_response = None
 
     @property
     def is_active(self):

google/api_core/exceptions.py

@@ -517,14 +517,14 @@ def format_http_response_error(
     errors = payload.get("error", {}).get("errors", ())
     # In JSON, details are already formatted in developer-friendly way.
     details = payload.get("error", {}).get("details", ())
-    error_info = list(
+    error_info_list = list(
         filter(
             lambda detail: detail.get("@type", "")
             == "type.googleapis.com/google.rpc.ErrorInfo",
             details,
         )
     )
-    error_info = error_info[0] if error_info else None
+    error_info = error_info_list[0] if error_info_list else None
     message = _format_rest_error_message(error_message, method, url)
 
     exception = from_http_status(

google/api_core/grpc_helpers_async.py

@@ -152,9 +152,9 @@ class _WrappedStreamStreamCall(
 
 
 # public type alias denoting the return type of async streaming gapic calls
-GrpcAsyncStream = _WrappedStreamResponseMixin[P]
+GrpcAsyncStream = _WrappedStreamResponseMixin
 # public type alias denoting the return type of unary gapic calls
-AwaitableGrpcCall = _WrappedUnaryResponseMixin[P]
+AwaitableGrpcCall = _WrappedUnaryResponseMixin
 
 
 def _wrap_unary_errors(callable_):

google/api_core/retry/retry_unary.py

@@ -83,7 +83,7 @@ def check_if_exists():
 
 
 def retry_target(
-    target: Callable[_P, _R],
+    target: Callable[[], _R],
     predicate: Callable[[Exception], bool],
     sleep_generator: Iterable[float],
     timeout: float | None = None,

google/api_core/retry/retry_unary_async.py

@@ -94,7 +94,7 @@ async def check_if_exists():
 
 
 async def retry_target(
-    target: Callable[_P, Awaitable[_R]],
+    target: Callable[[], Awaitable[_R]],
     predicate: Callable[[Exception], bool],
     sleep_generator: Iterable[float],
     timeout: float | None = None,

noxfile.py

@@ -275,9 +275,7 @@ def pytype(session):
 @nox.session(python=DEFAULT_PYTHON_VERSION)
 def mypy(session):
     """Run type-checking."""
-    # TODO(https://github.com/googleapis/python-api-core/issues/682):
-    # Use the latest version of mypy instead of mypy<1.11.0
-    session.install(".[grpc,async_rest]", "mypy<1.11.0")
+    session.install(".[grpc,async_rest]", "mypy")
     session.install(
         "types-setuptools",
         "types-requests",

tests/asyncio/test_grpc_helpers_async.py

@@ -319,7 +319,7 @@ def test_awaitable_grpc_call():
     """
     AwaitableGrpcCall type should be an Awaitable and a grpc.aio.Call.
     """
-    instance = grpc_helpers_async.AwaitableGrpcCall[int]()
+    instance = grpc_helpers_async.AwaitableGrpcCall()
     assert isinstance(instance, grpc.aio.Call)
     # should implement __await__
     assert hasattr(instance, "__await__")

tests/unit/test_bidi.py

@@ -296,6 +296,9 @@ def test_close(self):
         # ensure the request queue was signaled to stop.
         assert bidi_rpc.pending_requests == 1
         assert bidi_rpc._request_queue.get() is None
+        # ensure request and callbacks are cleaned up
+        assert bidi_rpc._initial_request is None
+        assert not bidi_rpc._callbacks
 
     def test_close_no_rpc(self):
         bidi_rpc = bidi.BidiRpc(None)
@@ -623,6 +626,8 @@ def cancel_side_effect():
         assert bidi_rpc.pending_requests == 1
         assert bidi_rpc._request_queue.get() is None
         assert bidi_rpc._finalized
+        assert bidi_rpc._initial_request is None
+        assert not bidi_rpc._callbacks
 
     def test_reopen_failure_on_rpc_restart(self):
         error1 = ValueError("1")
@@ -777,6 +782,7 @@ def on_response(response):
         consumer.stop()
 
         assert consumer.is_active is False
+        assert consumer._on_response is None
 
     def test_wake_on_error(self):
         should_continue = threading.Event()
@@ -884,6 +890,7 @@ def close_side_effect():
 
         consumer.stop()
         assert consumer.is_active is False
+        assert consumer._on_response is None
 
         # calling stop twice should not result in an error.
         consumer.stop()