From b9caaf8a63bc3280645dca2788a8c4ed1a556769 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Fri, 16 May 2025 17:13:52 +0500 Subject: [PATCH] Simplify deferred_from_coro(), add more tests. --- scrapy/utils/defer.py | 34 +++--- tests/test_utils_defer.py | 219 ++++++++++++++++++++++++++++++++++---- 2 files changed, 212 insertions(+), 41 deletions(-) diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index 6e1687f3e..d06397f50 100644 --- a/scrapy/utils/defer.py +++ b/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]: diff --git a/tests/test_utils_defer.py b/tests/test_utils_defer.py index 36bd8ced9..29cd5fbf2 100644 --- a/tests/test_utils_defer.py +++ b/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