-
Notifications
You must be signed in to change notification settings - Fork 10.4k
/
defer.py
354 lines (295 loc) · 13.4 KB
/
defer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
"""
Helper functions for dealing with Twisted deferreds
"""
import asyncio
import inspect
from asyncio import Future
from functools import wraps
from typing import (
Any,
AsyncGenerator,
AsyncIterable,
Callable,
Coroutine,
Generator,
Iterable,
Iterator,
List,
Optional,
Union
)
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
from scrapy.exceptions import IgnoreRequest
from scrapy.utils.reactor import is_asyncio_reactor_installed
def defer_fail(_failure: Failure) -> Deferred:
"""Same as twisted.internet.defer.fail but delay calling errback until
next reactor loop
It delays by 100ms so reactor has a chance to go through readers and writers
before attending pending delayed calls, so do not set delay to zero.
"""
from twisted.internet import reactor
d = Deferred()
reactor.callLater(0.1, d.errback, _failure)
return d
def defer_succeed(result) -> Deferred:
"""Same as twisted.internet.defer.succeed but delay calling callback until
next reactor loop
It delays by 100ms so reactor has a chance to go through readers and writers
before attending pending delayed calls, so do not set delay to zero.
"""
from twisted.internet import reactor
d = Deferred()
reactor.callLater(0.1, d.callback, result)
return d
def defer_result(result) -> Deferred:
if isinstance(result, Deferred):
return result
elif isinstance(result, failure.Failure):
return defer_fail(result)
else:
return defer_succeed(result)
def mustbe_deferred(f: Callable, *args, **kw) -> Deferred:
"""Same as twisted.internet.defer.maybeDeferred, but delay calling
callback/errback to next reactor loop
"""
try:
result = f(*args, **kw)
# FIXME: Hack to avoid introspecting tracebacks. This to speed up
# processing of IgnoreRequest errors which are, by far, the most common
# exception in Scrapy - see #125
except IgnoreRequest as e:
return defer_fail(failure.Failure(e))
except Exception:
return defer_fail(failure.Failure())
else:
return defer_result(result)
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 = Cooperator()
work = (callable(elem, *args, **named) for elem in iterable)
return DeferredList([coop.coiterate(work) for _ in range(count)])
class _AsyncCooperatorAdapter(Iterator):
""" A class that wraps an async iterable into a normal iterator suitable
for using in Cooperator.coiterate(). As it's only needed for parallel_async(),
it calls the callable directly in the callback, instead of providing a more
generic interface.
On the outside, this class behaves as an iterator that yields Deferreds.
Each Deferred is fired with the result of the callable which was called on
the next result from aiterator. It raises StopIteration when aiterator is
exhausted, as expected.
Cooperator calls __next__() multiple times and waits on the Deferreds
returned from it. As async generators (since Python 3.8) don't support
awaiting on __anext__() several times in parallel, we need to serialize
this. It's done by storing the Deferreds returned from __next__() and
firing the oldest one when a result from __anext__() is available.
The workflow:
1. When __next__() is called for the first time, it creates a Deferred, stores it
in self.waiting_deferreds and returns it. It also makes a Deferred that will wait
for self.aiterator.__anext__() and puts it into self.anext_deferred.
2. If __next__() is called again before self.anext_deferred fires, more Deferreds
are added to self.waiting_deferreds.
3. When self.anext_deferred fires, it either calls _callback() or _errback(). Both
clear self.anext_deferred.
3.1. _callback() calls the callable passing the result value that it takes, pops a
Deferred from self.waiting_deferreds, and if the callable result was a Deferred, it
chains those Deferreds so that the waiting Deferred will fire when the result
Deferred does, otherwise it fires it directly. This causes one awaiting task to
receive a result. If self.waiting_deferreds is still not empty, new __anext__() is
called and self.anext_deferred is populated.
3.2. _errback() checks the exception class. If it's StopAsyncIteration it means
self.aiterator is exhausted and so it sets self.finished and fires all
self.waiting_deferreds. Other exceptions are propagated.
4. If __next__() is called after __anext__() was handled, then if self.finished is
True, it raises StopIteration, otherwise it acts like in step 2, but if
self.anext_deferred is now empty is also populates it with a new __anext__().
Note that CooperativeTask ignores the value returned from the Deferred that it waits
for, so we fire them with None when needed.
It may be possible to write an async iterator-aware replacement for
Cooperator/CooperativeTask and use it instead of this adapter to achieve the same
goal.
"""
def __init__(self, aiterable: AsyncIterable, callable: Callable, *callable_args, **callable_kwargs):
self.aiterator = aiterable.__aiter__()
self.callable = callable
self.callable_args = callable_args
self.callable_kwargs = callable_kwargs
self.finished = False
self.waiting_deferreds: List[Deferred] = []
self.anext_deferred: Optional[Deferred] = None
def _callback(self, result: Any) -> None:
# This gets called when the result from aiterator.__anext__() is available.
# It calls the callable on it and sends the result to the oldest waiting Deferred
# (by chaining if the result is a Deferred too or by firing if not).
self.anext_deferred = None
result = self.callable(result, *self.callable_args, **self.callable_kwargs)
d = self.waiting_deferreds.pop(0)
if isinstance(result, Deferred):
result.chainDeferred(d)
else:
d.callback(None)
if self.waiting_deferreds:
self._call_anext()
def _errback(self, failure: Failure) -> None:
# This gets called on any exceptions in aiterator.__anext__().
# It handles StopAsyncIteration by stopping the iteration and reraises all others.
self.anext_deferred = None
failure.trap(StopAsyncIteration)
self.finished = True
for d in self.waiting_deferreds:
d.callback(None)
def _call_anext(self) -> None:
# This starts waiting for the next result from aiterator.
# If aiterator is exhausted, _errback will be called.
self.anext_deferred = deferred_from_coro(self.aiterator.__anext__())
self.anext_deferred.addCallbacks(self._callback, self._errback)
def __next__(self) -> Deferred:
# This puts a new Deferred into self.waiting_deferreds and returns it.
# It also calls __anext__() if needed.
if self.finished:
raise StopIteration
d = Deferred()
self.waiting_deferreds.append(d)
if not self.anext_deferred:
self._call_anext()
return d
def parallel_async(async_iterable: AsyncIterable, count: int, callable: Callable, *args, **named) -> DeferredList:
""" Like parallel but for async iterators """
coop = Cooperator()
work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named)
dl = DeferredList([coop.coiterate(work) for _ in range(count)])
return dl
def process_chain(callbacks: Iterable[Callable], input, *a, **kw) -> Deferred:
"""Return a Deferred built by chaining the given callbacks"""
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) -> Deferred:
"""Return a Deferred built by chaining the given callbacks and errbacks"""
d = Deferred()
for cb, eb in zip(callbacks, errbacks):
d.addCallbacks(
callback=cb, errback=eb,
callbackArgs=a, callbackKeywords=kw,
errbackArgs=a, errbackKeywords=kw,
)
if isinstance(input, failure.Failure):
d.errback(input)
else:
d.callback(input)
return d
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 = DeferredList(dfds, fireOnOneErrback=True, consumeErrors=True)
d.addCallbacks(lambda r: [x[1] for x in r], lambda f: f.value.subFailure)
return d
def iter_errback(iterable: Iterable, errback: Callable, *a, **kw) -> Generator:
"""Wraps an iterable calling an errback if an error is caught while
iterating it.
"""
it = iter(iterable)
while True:
try:
yield next(it)
except StopIteration:
break
except Exception:
errback(failure.Failure(), *a, **kw)
async def aiter_errback(aiterable: AsyncIterable, errback: Callable, *a, **kw) -> AsyncGenerator:
"""Wraps an async iterable calling an errback if an error is caught while
iterating it. Similar to scrapy.utils.defer.iter_errback()
"""
it = aiterable.__aiter__()
while True:
try:
yield await it.__anext__()
except StopAsyncIteration:
break
except Exception:
errback(failure.Failure(), *a, **kw)
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, 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 ensureDeferred(o)
else:
# wrapping the coroutine into a Future and then into a Deferred, this requires AsyncioSelectorReactor
return Deferred.fromFuture(asyncio.ensure_future(o))
return o
def deferred_f_from_coro_f(coro_f: Callable[..., Coroutine]) -> Callable:
""" Converts a coroutine function into a function that returns a Deferred.
The coroutine function will be called at the time when the wrapper is called. Wrapper args will be passed to it.
This is useful for callback chains, as callback functions are called with the previous callback result.
"""
@wraps(coro_f)
def f(*coro_args, **coro_kwargs):
return deferred_from_coro(coro_f(*coro_args, **coro_kwargs))
return f
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=Deferred.debug))
if isinstance(result, Deferred):
return result
elif asyncio.isfuture(result) or inspect.isawaitable(result):
return deferred_from_coro(result)
elif isinstance(result, failure.Failure):
return defer.fail(result)
else:
return defer.succeed(result)
def deferred_to_future(d: Deferred) -> Future:
"""
.. versionadded:: 2.6.0
Return an :class:`asyncio.Future` object that wraps *d*.
When :ref:`using the asyncio reactor <install-asyncio>`, you cannot await
on :class:`~twisted.internet.defer.Deferred` objects from :ref:`Scrapy
callables defined as coroutines <coroutine-support>`, you can only await on
``Future`` objects. Wrapping ``Deferred`` objects into ``Future`` objects
allows you to wait on them::
class MySpider(Spider):
...
async def parse(self, response):
d = treq.get('https://example.com/additional')
additional_response = await deferred_to_future(d)
"""
return d.asFuture(asyncio.get_event_loop())
def maybe_deferred_to_future(d: Deferred) -> Union[Deferred, Future]:
"""
.. versionadded:: 2.6.0
Return *d* as an object that can be awaited from a :ref:`Scrapy callable
defined as a coroutine <coroutine-support>`.
What you can await in Scrapy callables defined as coroutines depends on the
value of :setting:`TWISTED_REACTOR`:
- When not using the asyncio reactor, you can only await on
:class:`~twisted.internet.defer.Deferred` objects.
- When :ref:`using the asyncio reactor <install-asyncio>`, you can only
await on :class:`asyncio.Future` objects.
If you want to write code that uses ``Deferred`` objects but works with any
reactor, use this function on all ``Deferred`` objects::
class MySpider(Spider):
...
async def parse(self, response):
d = treq.get('https://example.com/additional')
extra_response = await maybe_deferred_to_future(d)
"""
if not is_asyncio_reactor_installed():
return d
else:
return deferred_to_future(d)