Simplify deferred_from_coro(), add more tests.

scrapy/utils/defer.py

@@ -10,14 +10,13 @@ import warnings
 from asyncio import Future
 from collections.abc import Awaitable, Coroutine, Iterable, Iterator
 from functools import wraps
-from types import CoroutineType
-from typing import TYPE_CHECKING, Any, Generic, TypeVar, Union, cast, overload
+from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast, overload
 
-from twisted.internet import defer
 from twisted.internet.defer import (
     Deferred,
     DeferredList,
-    ensureDeferred,
+    fail,
+    succeed,
 )
 from twisted.internet.task import Cooperator
 from twisted.python import failure
@@ -315,7 +314,7 @@ def process_parallel(
     """Return a Deferred with the output of all successful calls to the given
     callbacks
     """
-    dfds = [defer.succeed(input).addCallback(x, *a, **kw) for x in callbacks]
+    dfds = [succeed(input).addCallback(x, *a, **kw) for x in callbacks]
     d: Deferred[list[tuple[bool, _T2]]] = DeferredList(
         dfds, fireOnOneErrback=True, consumeErrors=True
     )
@@ -366,27 +365,24 @@ async def aiter_errback(
             errback(failure.Failure(), *a, **kw)
 
 
-_CT = TypeVar("_CT", bound=Union[Awaitable, CoroutineType, Future])
+@overload
+def deferred_from_coro(o: Awaitable[_T]) -> Deferred[_T]: ...
 
 
 @overload
-def deferred_from_coro(o: _CT) -> Deferred: ...
+def deferred_from_coro(o: _T2) -> _T2: ...
 
 
-@overload
-def deferred_from_coro(o: _T) -> _T: ...
-
-
-def deferred_from_coro(o: _T) -> Deferred | _T:
+def deferred_from_coro(o: Awaitable[_T] | _T2) -> Deferred[_T] | _T2:
     """Converts a coroutine or other awaitable object into a Deferred,
     or returns the object as is if it isn't a coroutine."""
     if isinstance(o, Deferred):
         return o
-    if asyncio.isfuture(o) or inspect.isawaitable(o):
+    if inspect.isawaitable(o):
         if not is_asyncio_reactor_installed():
             # wrapping the coroutine directly into a Deferred, this doesn't work correctly with coroutines
             # that use asyncio, e.g. "await asyncio.sleep(1)"
-            return ensureDeferred(cast(Coroutine[Deferred, Any, Any], o))
+            return Deferred.fromCoroutine(cast(Coroutine[Deferred[Any], Any, _T], o))
         # wrapping the coroutine into a Future and then into a Deferred, this requires AsyncioSelectorReactor
         event_loop = _get_asyncio_event_loop()
         return Deferred.fromFuture(asyncio.ensure_future(o, loop=event_loop))
@@ -394,7 +390,7 @@ def deferred_from_coro(o: _T) -> Deferred | _T:
 
 
 def deferred_f_from_coro_f(
-    coro_f: Callable[_P, Coroutine[Any, Any, _T]],
+    coro_f: Callable[_P, Awaitable[_T]],
 ) -> Callable[_P, Deferred[_T]]:
     """Converts a coroutine function into a function that returns a Deferred.
 
@@ -403,7 +399,7 @@ def deferred_f_from_coro_f(
     """
 
     @wraps(coro_f)
-    def f(*coro_args: _P.args, **coro_kwargs: _P.kwargs) -> Any:
+    def f(*coro_args: _P.args, **coro_kwargs: _P.kwargs) -> Deferred[_T]:
         return deferred_from_coro(coro_f(*coro_args, **coro_kwargs))
 
     return f
@@ -416,15 +412,15 @@ def maybeDeferred_coro(
     try:
         result = f(*args, **kw)
     except:  # noqa: E722  # pylint: disable=bare-except
-        return defer.fail(failure.Failure(captureVars=Deferred.debug))
+        return fail(failure.Failure(captureVars=Deferred.debug))
 
     if isinstance(result, Deferred):
         return result
     if asyncio.isfuture(result) or inspect.isawaitable(result):
         return deferred_from_coro(result)
     if isinstance(result, failure.Failure):
-        return defer.fail(result)
-    return defer.succeed(result)
+        return fail(result)
+    return succeed(result)
 
 
 def deferred_to_future(d: Deferred[_T]) -> Future[_T]:

tests/test_utils_defer.py

@@ -1,7 +1,12 @@
+from __future__ import annotations
+
+import asyncio
 import random
+from asyncio import Future
+from typing import TYPE_CHECKING, Any
 
 import pytest
-from twisted.internet import defer, reactor
+from twisted.internet.defer import Deferred, inlineCallbacks, succeed
 from twisted.python.failure import Failure
 from twisted.trial import unittest
 
@@ -9,6 +14,8 @@ from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen
 from scrapy.utils.defer import (
     aiter_errback,
     deferred_f_from_coro_f,
+    deferred_from_coro,
+    deferred_to_future,
     iter_errback,
     maybe_deferred_to_future,
     mustbe_deferred,
@@ -17,12 +24,15 @@ from scrapy.utils.defer import (
     process_parallel,
 )
 
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
+
 
 class TestMustbeDeferred(unittest.TestCase):
-    def test_success_function(self):
-        steps = []
+    def test_success_function(self) -> Deferred[list[int]]:
+        steps: list[int] = []
 
-        def _append(v):
+        def _append(v: int) -> list[int]:
             steps.append(v)
             return steps
 
@@ -31,12 +41,14 @@ class TestMustbeDeferred(unittest.TestCase):
         steps.append(2)  # add another value, that should be caught by assertEqual
         return dfd
 
-    def test_unfired_deferred(self):
-        steps = []
+    def test_unfired_deferred(self) -> Deferred[list[int]]:
+        steps: list[int] = []
+
+        def _append(v: int) -> Deferred[list[int]]:
+            from twisted.internet import reactor
 
-        def _append(v):
             steps.append(v)
-            dfd = defer.Deferred()
+            dfd: Deferred[list[int]] = Deferred()
             reactor.callLater(0, dfd.callback, steps)
             return dfd
 
@@ -51,7 +63,7 @@ def cb1(value, arg1, arg2):
 
 
 def cb2(value, arg1, arg2):
-    return defer.succeed(f"(cb2 {value} {arg1} {arg2})")
+    return succeed(f"(cb2 {value} {arg1} {arg2})")
 
 
 def cb3(value, arg1, arg2):
@@ -67,7 +79,7 @@ def eb1(failure, arg1, arg2):
 
 
 class TestDeferUtils(unittest.TestCase):
-    @defer.inlineCallbacks
+    @inlineCallbacks
     def test_process_chain(self):
         x = yield process_chain([cb1, cb2, cb3], "res", "v1", "v2")
         assert x == "(cb3 (cb2 (cb1 res v1 v2) v1 v2) v1 v2)"
@@ -75,7 +87,7 @@ class TestDeferUtils(unittest.TestCase):
         with pytest.raises(TypeError):
             yield process_chain([cb1, cb_fail, cb3], "res", "v1", "v2")
 
-    @defer.inlineCallbacks
+    @inlineCallbacks
     def test_process_parallel(self):
         x = yield process_parallel([cb1, cb2, cb3], "res", "v1", "v2")
         assert x == ["(cb1 res v1 v2)", "(cb2 res v1 v2)", "(cb3 res v1 v2)"]
@@ -88,7 +100,7 @@ class TestDeferUtils(unittest.TestCase):
 
 class TestIterErrback:
     def test_iter_errback_good(self):
-        def itergood():
+        def itergood() -> Generator[int, None, None]:
             yield from range(10)
 
         errors = []
@@ -97,7 +109,7 @@ class TestIterErrback:
         assert not errors
 
     def test_iter_errback_bad(self):
-        def iterbad():
+        def iterbad() -> Generator[int, None, None]:
             for x in range(10):
                 if x == 5:
                     1 / 0
@@ -113,7 +125,7 @@ class TestIterErrback:
 class TestAiterErrback(unittest.TestCase):
     @deferred_f_from_coro_f
     async def test_aiter_errback_good(self):
-        async def itergood():
+        async def itergood() -> AsyncGenerator[int, None]:
             for x in range(10):
                 yield x
 
@@ -124,7 +136,7 @@ class TestAiterErrback(unittest.TestCase):
 
     @deferred_f_from_coro_f
     async def test_iter_errback_bad(self):
-        async def iterbad():
+        async def iterbad() -> AsyncGenerator[int, None]:
             for x in range(10):
                 if x == 5:
                     1 / 0
@@ -168,10 +180,12 @@ class TestAsyncCooperator(unittest.TestCase):
     CONCURRENT_ITEMS = 50
 
     @staticmethod
-    def callable(o, results):
+    def callable(o: int, results: list[int]) -> Deferred[None] | None:
+        from twisted.internet import reactor
+
         if random.random() < 0.4:
             # simulate async processing
-            dfd = defer.Deferred()
+            dfd: Deferred[None] = Deferred()
             dfd.addCallback(lambda _: results.append(o))
             delay = random.random() / 8
             reactor.callLater(delay, dfd.callback, None)
@@ -181,22 +195,24 @@ class TestAsyncCooperator(unittest.TestCase):
         return None
 
     @staticmethod
-    def get_async_iterable(length):
+    def get_async_iterable(length: int) -> AsyncGenerator[int, None]:
         # simulate a simple callback without delays between results
         return as_async_generator(range(length))
 
     @staticmethod
-    async def get_async_iterable_with_delays(length):
+    async def get_async_iterable_with_delays(length: int) -> AsyncGenerator[int, None]:
         # simulate a callback with delays between some of the results
+        from twisted.internet import reactor
+
         for i in range(length):
             if random.random() < 0.1:
-                dfd = defer.Deferred()
+                dfd: Deferred[None] = Deferred()
                 delay = random.random() / 20
                 reactor.callLater(delay, dfd.callback, None)
                 await maybe_deferred_to_future(dfd)
             yield i
 
-    @defer.inlineCallbacks
+    @inlineCallbacks
     def test_simple(self):
         for length in [20, 50, 100]:
             results = []
@@ -205,7 +221,7 @@ class TestAsyncCooperator(unittest.TestCase):
             yield dl
             assert list(range(length)) == sorted(results)
 
-    @defer.inlineCallbacks
+    @inlineCallbacks
     def test_delays(self):
         for length in [20, 50, 100]:
             results = []
@@ -213,3 +229,162 @@ class TestAsyncCooperator(unittest.TestCase):
             dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
             yield dl
             assert list(range(length)) == sorted(results)
+
+
+class TestDeferredFromCoro(unittest.TestCase):
+    def test_deferred(self):
+        d = Deferred()
+        result = deferred_from_coro(d)
+        assert isinstance(result, Deferred)
+        assert result is d
+
+    def test_object(self):
+        result = deferred_from_coro(42)
+        assert result == 42
+
+    @inlineCallbacks
+    def test_coroutine(self):
+        async def coroutine() -> int:
+            return 42
+
+        result = deferred_from_coro(coroutine())
+        assert isinstance(result, Deferred)
+        coro_result = yield result
+        assert coro_result == 42
+
+    @pytest.mark.only_asyncio
+    @inlineCallbacks
+    def test_coroutine_asyncio(self):
+        async def coroutine() -> int:
+            await asyncio.sleep(0)
+            return 42
+
+        result = deferred_from_coro(coroutine())
+        assert isinstance(result, Deferred)
+        coro_result = yield result
+        assert coro_result == 42
+
+    @pytest.mark.only_asyncio
+    @inlineCallbacks
+    def test_future(self):
+        future = Future()
+        result = deferred_from_coro(future)
+        assert isinstance(result, Deferred)
+        future.set_result(42)
+        future_result = yield result
+        assert future_result == 42
+
+
+class TestDeferredFFromCoroF(unittest.TestCase):
+    @inlineCallbacks
+    def _assert_result(
+        self, c_f: Callable[[], Awaitable[int]]
+    ) -> Generator[Deferred[Any], Any, None]:
+        d_f = deferred_f_from_coro_f(c_f)
+        d = d_f()
+        assert isinstance(d, Deferred)
+        result = yield d
+        assert result == 42
+
+    @inlineCallbacks
+    def test_coroutine(self):
+        async def c_f() -> int:
+            return 42
+
+        yield self._assert_result(c_f)
+
+    @inlineCallbacks
+    def test_coroutine_asyncio(self):
+        async def c_f() -> int:
+            return 42
+
+        yield self._assert_result(c_f)
+
+    @pytest.mark.only_asyncio
+    @inlineCallbacks
+    def test_future(self):
+        def c_f() -> Future[int]:
+            f: Future[int] = Future()
+            f.set_result(42)
+            return f
+
+        yield self._assert_result(c_f)
+
+
+class TestDeferredToFuture(unittest.TestCase):
+    @deferred_f_from_coro_f
+    async def test_deferred(self):
+        d = Deferred()
+        result = deferred_to_future(d)
+        assert isinstance(result, Future)
+        d.callback(42)
+        future_result = await result
+        assert future_result == 42
+
+    @deferred_f_from_coro_f
+    async def test_wrapped_coroutine(self):
+        async def c_f() -> int:
+            return 42
+
+        d = deferred_from_coro(c_f())
+        result = deferred_to_future(d)
+        assert isinstance(result, Future)
+        future_result = await result
+        assert future_result == 42
+
+    @pytest.mark.only_asyncio
+    @deferred_f_from_coro_f
+    async def test_wrapped_coroutine_asyncio(self):
+        async def c_f() -> int:
+            await asyncio.sleep(0)
+            return 42
+
+        d = deferred_from_coro(c_f())
+        result = maybe_deferred_to_future(d)
+        assert isinstance(result, Future)
+        future_result = await result
+        assert future_result == 42
+
+
+@pytest.mark.only_asyncio
+class TestMaybeDeferredToFutureAsyncio(unittest.TestCase):
+    @deferred_f_from_coro_f
+    async def test_deferred(self):
+        d = Deferred()
+        result = maybe_deferred_to_future(d)
+        assert isinstance(result, Future)
+        d.callback(42)
+        future_result = await result
+        assert future_result == 42
+
+    @deferred_f_from_coro_f
+    async def test_wrapped_coroutine(self):
+        async def c_f() -> int:
+            return 42
+
+        d = deferred_from_coro(c_f())
+        result = maybe_deferred_to_future(d)
+        assert isinstance(result, Future)
+        future_result = await result
+        assert future_result == 42
+
+    @deferred_f_from_coro_f
+    async def test_wrapped_coroutine_asyncio(self):
+        async def c_f() -> int:
+            await asyncio.sleep(0)
+            return 42
+
+        d = deferred_from_coro(c_f())
+        result = maybe_deferred_to_future(d)
+        assert isinstance(result, Future)
+        future_result = await result
+        assert future_result == 42
+
+
+@pytest.mark.only_not_asyncio
+class TestMaybeDeferredToFutureNotAsyncio:
+    def test_deferred(self):
+        d = Deferred()
+        result = maybe_deferred_to_future(d)
+        assert isinstance(result, Deferred)
+        assert result is d