From 08e4eaf97369ba6daa4b5d84e00fd4d36b78e00a Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Tue, 13 Apr 2021 22:41:01 +0500 Subject: [PATCH] Import Deferred directly in scrapy/utils/defer.py. --- scrapy/utils/defer.py | 48 ++++++++++++++++++++++--------------------- 1 file changed, 25 insertions(+), 23 deletions(-) diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index e1139b1d1..b317c12a3 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -7,7 +7,9 @@ from collections.abc import Coroutine from functools import wraps from typing import Any, Callable, Generator, Iterable -from twisted.internet import defer, task +from twisted.internet import defer +from twisted.internet.defer import Deferred, DeferredList, ensureDeferred +from twisted.internet.task import Cooperator from twisted.python import failure from twisted.python.failure import Failure @@ -15,7 +17,7 @@ from scrapy.exceptions import IgnoreRequest from scrapy.utils.reactor import is_asyncio_reactor_installed -def defer_fail(_failure: Failure) -> defer.Deferred: +def defer_fail(_failure: Failure) -> Deferred: """Same as twisted.internet.defer.fail but delay calling errback until next reactor loop @@ -23,12 +25,12 @@ def defer_fail(_failure: Failure) -> defer.Deferred: before attending pending delayed calls, so do not set delay to zero. """ from twisted.internet import reactor - d = defer.Deferred() + d = Deferred() reactor.callLater(0.1, d.errback, _failure) return d -def defer_succeed(result) -> defer.Deferred: +def defer_succeed(result) -> Deferred: """Same as twisted.internet.defer.succeed but delay calling callback until next reactor loop @@ -36,13 +38,13 @@ def defer_succeed(result) -> defer.Deferred: before attending pending delayed calls, so do not set delay to zero. """ from twisted.internet import reactor - d = defer.Deferred() + d = Deferred() reactor.callLater(0.1, d.callback, result) return d -def defer_result(result) -> defer.Deferred: - if isinstance(result, defer.Deferred): +def defer_result(result) -> Deferred: + if isinstance(result, Deferred): return result elif isinstance(result, failure.Failure): return defer_fail(result) @@ -50,7 +52,7 @@ def defer_result(result) -> defer.Deferred: return defer_succeed(result) -def mustbe_deferred(f: Callable, *args, **kw) -> defer.Deferred: +def mustbe_deferred(f: Callable, *args, **kw) -> Deferred: """Same as twisted.internet.defer.maybeDeferred, but delay calling callback/errback to next reactor loop """ @@ -67,29 +69,29 @@ def mustbe_deferred(f: Callable, *args, **kw) -> defer.Deferred: return defer_result(result) -def parallel(iterable: Iterable, count: int, callable: Callable, *args, **named) -> defer.DeferredList: +def parallel(iterable: Iterable, count: int, callable: Callable, *args, **named) -> DeferredList: """Execute a callable over the objects in the given iterable, in parallel, using no more than ``count`` concurrent calls. Taken from: https://jcalderone.livejournal.com/24285.html """ - coop = task.Cooperator() + coop = Cooperator() work = (callable(elem, *args, **named) for elem in iterable) - return defer.DeferredList([coop.coiterate(work) for _ in range(count)]) + return DeferredList([coop.coiterate(work) for _ in range(count)]) -def process_chain(callbacks: Iterable[Callable], input, *a, **kw) -> defer.Deferred: +def process_chain(callbacks: Iterable[Callable], input, *a, **kw) -> Deferred: """Return a Deferred built by chaining the given callbacks""" - d = defer.Deferred() + d = Deferred() for x in callbacks: d.addCallback(x, *a, **kw) d.callback(input) return d -def process_chain_both(callbacks: Iterable[Callable], errbacks: Iterable[Callable], input, *a, **kw) -> defer.Deferred: +def process_chain_both(callbacks: Iterable[Callable], errbacks: Iterable[Callable], input, *a, **kw) -> Deferred: """Return a Deferred built by chaining the given callbacks and errbacks""" - d = defer.Deferred() + d = Deferred() for cb, eb in zip(callbacks, errbacks): d.addCallbacks( callback=cb, errback=eb, @@ -103,12 +105,12 @@ def process_chain_both(callbacks: Iterable[Callable], errbacks: Iterable[Callabl return d -def process_parallel(callbacks: Iterable[Callable], input, *a, **kw) -> defer.Deferred: +def process_parallel(callbacks: Iterable[Callable], input, *a, **kw) -> Deferred: """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] - d = defer.DeferredList(dfds, fireOnOneErrback=True, consumeErrors=True) + d = DeferredList(dfds, fireOnOneErrback=True, consumeErrors=True) d.addCallbacks(lambda r: [x[1] for x in r], lambda f: f.value.subFailure) return d @@ -129,16 +131,16 @@ def iter_errback(iterable: Iterable, errback: Callable, *a, **kw) -> Generator: def deferred_from_coro(o) -> Any: """Converts a coroutine into a Deferred, or returns the object as is if it isn't a coroutine""" - if isinstance(o, defer.Deferred): + if isinstance(o, Deferred): return o if asyncio.isfuture(o) or 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 defer.ensureDeferred(o) + return ensureDeferred(o) else: # wrapping the coroutine into a Future and then into a Deferred, this requires AsyncioSelectorReactor - return defer.Deferred.fromFuture(asyncio.ensure_future(o)) + return Deferred.fromFuture(asyncio.ensure_future(o)) return o @@ -154,14 +156,14 @@ def deferred_f_from_coro_f(coro_f: Callable[..., Coroutine]) -> Callable: return f -def maybeDeferred_coro(f: Callable, *args, **kw) -> defer.Deferred: +def maybeDeferred_coro(f: Callable, *args, **kw) -> Deferred: """ Copy of defer.maybeDeferred that also converts coroutines to Deferreds. """ try: result = f(*args, **kw) except: # noqa: E722 - return defer.fail(failure.Failure(captureVars=defer.Deferred.debug)) + return defer.fail(failure.Failure(captureVars=Deferred.debug)) - if isinstance(result, defer.Deferred): + if isinstance(result, Deferred): return result elif asyncio.isfuture(result) or inspect.isawaitable(result): return deferred_from_coro(result)