fix generator close for py3.13

Author: daniel-sanche

google/cloud/firestore_v1/async_aggregation.py

@@ -88,12 +88,15 @@ async def get(
             timeout=timeout,
             explain_options=explain_options,
         )
-        result = [aggregation async for aggregation in stream_result]
-
-        if explain_options is None:
-            explain_metrics = None
-        else:
-            explain_metrics = await stream_result.get_explain_metrics()
+        try:
+            result = [aggregation async for aggregation in stream_result]
+
+            if explain_options is None:
+                explain_metrics = None
+            else:
+                explain_metrics = await stream_result.get_explain_metrics()
+        finally:
+            await stream_result.aclose()
 
         return QueryResultsList(result, explain_options, explain_metrics)
 

google/cloud/firestore_v1/async_query.py

@@ -231,14 +231,17 @@ async def get(
             timeout=timeout,
             explain_options=explain_options,
         )
-        result_list = [d async for d in result]
-        if is_limited_to_last:
-            result_list = list(reversed(result_list))
-
-        if explain_options is None:
-            explain_metrics = None
-        else:
-            explain_metrics = await result.get_explain_metrics()
+        try:
+            result_list = [d async for d in result]
+            if is_limited_to_last:
+                result_list = list(reversed(result_list))
+
+            if explain_options is None:
+                explain_metrics = None
+            else:
+                explain_metrics = await result.get_explain_metrics()
+        finally:
+            await result.aclose()
 
         return QueryResultsList(result_list, explain_options, explain_metrics)
 

google/cloud/firestore_v1/async_vector_query.py

@@ -91,12 +91,15 @@ async def get(
             timeout=timeout,
             explain_options=explain_options,
         )
-        result = [snapshot async for snapshot in stream_result]
+        try:
+            result = [snapshot async for snapshot in stream_result]
 
-        if explain_options is None:
-            explain_metrics = None
-        else:
-            explain_metrics = await stream_result.get_explain_metrics()
+            if explain_options is None:
+                explain_metrics = None
+            else:
+                explain_metrics = await stream_result.get_explain_metrics()
+        finally:
+            await stream_result.aclose()
 
         return QueryResultsList(result, explain_options, explain_metrics)
 
@@ -151,7 +154,6 @@ async def _make_stream(
             metadata=self._client._rpc_metadata,
             **kwargs,
         )
-
         async for response in response_iterator:
             if self._nested_query._all_descendants:
                 snapshot = _collection_group_query_response_to_snapshot(

tests/unit/v1/test__helpers.py

@@ -2538,17 +2538,12 @@ async def __call__(self, *args, **kwargs):
         return super(AsyncMock, self).__call__(*args, **kwargs)
 
 
-class AsyncIter:
+async def AsyncIter(items):
     """Utility to help recreate the effect of an async generator. Useful when
     you need to mock a system that requires `async for`.
     """
-
-    def __init__(self, items):
-        self.items = items
-
-    async def __aiter__(self):
-        for i in self.items:
-            yield i
+    for i in items:
+        yield i
 
 
 def _value_pb(**kwargs):

tests/unit/v1/test_async_aggregation.py

@@ -592,6 +592,7 @@ async def _async_aggregation_query_stream_helper(
             assert r.alias == aggregation_result.alias
             assert r.value == aggregation_result.value
         results.append(result)
+    await returned.aclose()
     assert len(results) == len(results_list)
 
     if explain_options is None:

tests/unit/v1/test_async_query.py

@@ -371,6 +371,7 @@ async def _stream_helper(retry=None, timeout=None, explain_options=None):
     assert isinstance(stream_response, AsyncStreamGenerator)
 
     returned = [x async for x in stream_response]
+    await stream_response.aclose()
     assert len(returned) == 1
     snapshot = returned[0]
     assert snapshot.reference._path == ("dee", "sleep")

tests/unit/v1/test_async_stream_generator.py

@@ -192,6 +192,7 @@ async def test_async_stream_generator_explain_metrics_explain_options_analyze_tr
         "index_entries_scanned": "index_entries_scanned",
     }
     assert actual_explain_metrics.execution_stats.debug_stats == expected_debug_stats
+    await inst.aclose()
 
 
 @pytest.mark.asyncio
@@ -230,6 +231,7 @@ async def test_async_stream_generator_explain_metrics_explain_options_analyze_fa
             }
         }
     ]
+    await inst.aclose()
 
 
 @pytest.mark.asyncio
@@ -242,6 +244,7 @@ async def test_async_stream_generator_explain_metrics_missing_explain_options_an
         query_profile.QueryExplainError, match="Did not receive explain_metrics"
     ):
         await inst.get_explain_metrics()
+    await inst.aclose()
 
 
 @pytest.mark.asyncio

tests/unit/v1/test_async_transaction.py

@@ -442,6 +442,7 @@ async def _get_w_query_helper(retry=None, timeout=None, explain_options=None):
         metadata=client._rpc_metadata,
         **kwargs,
     )
+    await returned_generator.aclose()
 
 
 @pytest.mark.asyncio

tests/unit/v1/test_async_vector_query.py

@@ -595,6 +595,7 @@ async def _async_vector_query_stream_helper(
     assert isinstance(returned, AsyncStreamGenerator)
 
     results_list = [item async for item in returned]
+    await returned.aclose()
     assert len(results_list) == 1
     assert results_list[0].to_dict() == data