_AsyncConcurrentMapIterable: safely retrive and store current event loop in new event_loop attribute for performance and 3.14 compliance

see python/cpython#126353

Author: ebonnal

streamable/iterators.py

@@ -608,6 +608,12 @@ def __init__(
     ) -> None:
         super().__init__(iterator, buffersize, ordered)
         self.transformation = wrap_error(transformation, StopIteration)
+        self.event_loop: asyncio.AbstractEventLoop
+        try:
+            self.event_loop = asyncio.get_event_loop()
+        except RuntimeError:
+            self.event_loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(self.event_loop)
 
     async def _safe_transformation(
         self, elem: T
@@ -627,16 +633,16 @@ def _launch_task(
     ) -> "Future[Union[U, _RaisingIterator.ExceptionContainer]]":
         return cast(
             "Future[Union[U, _RaisingIterator.ExceptionContainer]]",
-            asyncio.get_event_loop().create_task(self._safe_transformation(elem)),
+            self.event_loop.create_task(self._safe_transformation(elem)),
         )
 
     def _future_result_collection(
         self,
     ) -> FutureResultCollection[Union[U, _RaisingIterator.ExceptionContainer]]:
         if self.ordered:
-            return FIFOAsyncFutureResultCollection()
+            return FIFOAsyncFutureResultCollection(self.event_loop)
         else:
-            return FDFOAsyncFutureResultCollection()
+            return FDFOAsyncFutureResultCollection(self.event_loop)
 
 
 class AsyncConcurrentMapIterator(_RaisingIterator[U]):

streamable/util/futuretools.py

@@ -3,7 +3,7 @@
 from collections import deque
 from concurrent.futures import Future
 from queue import Queue
-from typing import Deque, Iterator, Sized, Type, TypeVar
+from typing import Awaitable, Deque, Iterator, Sized, Type, TypeVar, cast
 
 T = TypeVar("T")
 
@@ -75,26 +75,31 @@ class FIFOAsyncFutureResultCollection(DequeFutureResultCollection[T]):
     First In First Out
     """
 
+    def __init__(self, event_loop: asyncio.AbstractEventLoop) -> None:
+        super().__init__()
+        self.event_loop = event_loop
+
     def __next__(self) -> T:
-        return asyncio.get_event_loop().run_until_complete(self._futures.popleft())  # type: ignore
+        return self.event_loop.run_until_complete(
+            cast(Awaitable[T], self._futures.popleft())
+        )
 
 
 class FDFOAsyncFutureResultCollection(CallbackFutureResultCollection[T]):
     """
     First Done First Out
     """
 
-    def __init__(self) -> None:
+    def __init__(self, event_loop: asyncio.AbstractEventLoop) -> None:
         super().__init__()
-        self._waiter: asyncio.futures.Future[T] = (
-            asyncio.get_event_loop().create_future()
-        )
+        self.event_loop = event_loop
+        self._waiter: asyncio.futures.Future[T] = self.event_loop.create_future()
 
     def _done_callback(self, future: "Future[T]") -> None:
         self._waiter.set_result(future.result())
 
     def __next__(self) -> T:
-        result = asyncio.get_event_loop().run_until_complete(self._waiter)
+        result = self.event_loop.run_until_complete(self._waiter)
         self._n_futures -= 1
-        self._waiter = asyncio.get_event_loop().create_future()
+        self._waiter = self.event_loop.create_future()
         return result

tests/test_iterators.py

@@ -1,6 +1,11 @@
+import asyncio
 import unittest
 
-from streamable.iterators import ObserveIterator, _OSConcurrentMapIterable
+from streamable.iterators import (
+    ObserveIterator,
+    _AsyncConcurrentMapIterable,
+    _OSConcurrentMapIterable,
+)
 
 
 class TestIterators(unittest.TestCase):