/
defer.py
2019 lines (1570 loc) · 69.9 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
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# -*- test-case-name: twisted.test.test_defer -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Support for results that aren't immediately available.
Maintainer: Glyph Lefkowitz
@var _NO_RESULT: The result used to represent the fact that there is no
result. B{Never ever ever use this as an actual result for a Deferred}. You
have been warned.
@var _CONTINUE: A marker left in L{Deferred.callback}s to indicate a Deferred
chain. Always accompanied by a Deferred instance in the args tuple pointing
at the Deferred which is chained to the Deferred which has this marker.
"""
from __future__ import division, absolute_import, print_function
import attr
import traceback
import types
import warnings
from sys import exc_info, version_info
from functools import wraps
from incremental import Version
# Twisted imports
from twisted.python.compat import cmp, comparable
from twisted.python import lockfile, failure
from twisted.logger import Logger
from twisted.python.deprecate import warnAboutFunction, deprecated
from twisted.python._oldstyle import _oldStyle
log = Logger()
class AlreadyCalledError(Exception):
pass
class CancelledError(Exception):
"""
This error is raised by default when a L{Deferred} is cancelled.
"""
class TimeoutError(Exception):
"""
This error is raised by default when a L{Deferred} times out.
"""
def logError(err):
"""
Log and return failure.
This method can be used as an errback that passes the failure on to the
next errback unmodified. Note that if this is the last errback, and the
deferred gets garbage collected after being this errback has been called,
the clean up code logs it again.
"""
log.failure(None, err)
return err
def succeed(result):
"""
Return a L{Deferred} that has already had C{.callback(result)} called.
This is useful when you're writing synchronous code to an
asynchronous interface: i.e., some code is calling you expecting a
L{Deferred} result, but you don't actually need to do anything
asynchronous. Just return C{defer.succeed(theResult)}.
See L{fail} for a version of this function that uses a failing
L{Deferred} rather than a successful one.
@param result: The result to give to the Deferred's 'callback'
method.
@rtype: L{Deferred}
"""
d = Deferred()
d.callback(result)
return d
def fail(result=None):
"""
Return a L{Deferred} that has already had C{.errback(result)} called.
See L{succeed}'s docstring for rationale.
@param result: The same argument that L{Deferred.errback} takes.
@raise NoCurrentExceptionError: If C{result} is L{None} but there is no
current exception state.
@rtype: L{Deferred}
"""
d = Deferred()
d.errback(result)
return d
def execute(callable, *args, **kw):
"""
Create a L{Deferred} from a callable and arguments.
Call the given function with the given arguments. Return a L{Deferred}
which has been fired with its callback as the result of that invocation
or its C{errback} with a L{Failure} for the exception thrown.
"""
try:
result = callable(*args, **kw)
except:
return fail()
else:
return succeed(result)
def maybeDeferred(f, *args, **kw):
"""
Invoke a function that may or may not return a L{Deferred}.
Call the given function with the given arguments. If the returned
object is a L{Deferred}, return it. If the returned object is a L{Failure},
wrap it with L{fail} and return it. Otherwise, wrap it in L{succeed} and
return it. If an exception is raised, convert it to a L{Failure}, wrap it
in L{fail}, and then return it.
@type f: Any callable
@param f: The callable to invoke
@param args: The arguments to pass to C{f}
@param kw: The keyword arguments to pass to C{f}
@rtype: L{Deferred}
@return: The result of the function call, wrapped in a L{Deferred} if
necessary.
"""
try:
result = f(*args, **kw)
except:
return fail(failure.Failure(captureVars=Deferred.debug))
if isinstance(result, Deferred):
return result
elif isinstance(result, failure.Failure):
return fail(result)
else:
return succeed(result)
@deprecated(Version('Twisted', 17, 1, 0),
replacement='twisted.internet.defer.Deferred.addTimeout')
def timeout(deferred):
deferred.errback(failure.Failure(TimeoutError("Callback timed out")))
def passthru(arg):
return arg
def setDebugging(on):
"""
Enable or disable L{Deferred} debugging.
When debugging is on, the call stacks from creation and invocation are
recorded, and added to any L{AlreadyCalledError}s we raise.
"""
Deferred.debug=bool(on)
def getDebugging():
"""
Determine whether L{Deferred} debugging is enabled.
"""
return Deferred.debug
# See module docstring.
_NO_RESULT = object()
_CONTINUE = object()
@_oldStyle
class Deferred:
"""
This is a callback which will be put off until later.
Why do we want this? Well, in cases where a function in a threaded
program would block until it gets a result, for Twisted it should
not block. Instead, it should return a L{Deferred}.
This can be implemented for protocols that run over the network by
writing an asynchronous protocol for L{twisted.internet}. For methods
that come from outside packages that are not under our control, we use
threads (see for example L{twisted.enterprise.adbapi}).
For more information about Deferreds, see doc/core/howto/defer.html or
U{http://twistedmatrix.com/documents/current/core/howto/defer.html}
When creating a Deferred, you may provide a canceller function, which
will be called by d.cancel() to let you do any clean-up necessary if the
user decides not to wait for the deferred to complete.
@ivar called: A flag which is C{False} until either C{callback} or
C{errback} is called and afterwards always C{True}.
@type called: L{bool}
@ivar paused: A counter of how many unmatched C{pause} calls have been made
on this instance.
@type paused: L{int}
@ivar _suppressAlreadyCalled: A flag used by the cancellation mechanism
which is C{True} if the Deferred has no canceller and has been
cancelled, C{False} otherwise. If C{True}, it can be expected that
C{callback} or C{errback} will eventually be called and the result
should be silently discarded.
@type _suppressAlreadyCalled: L{bool}
@ivar _runningCallbacks: A flag which is C{True} while this instance is
executing its callback chain, used to stop recursive execution of
L{_runCallbacks}
@type _runningCallbacks: L{bool}
@ivar _chainedTo: If this L{Deferred} is waiting for the result of another
L{Deferred}, this is a reference to the other Deferred. Otherwise,
L{None}.
"""
called = False
paused = False
_debugInfo = None
_suppressAlreadyCalled = False
# Are we currently running a user-installed callback? Meant to prevent
# recursive running of callbacks when a reentrant call to add a callback is
# used.
_runningCallbacks = False
# Keep this class attribute for now, for compatibility with code that
# sets it directly.
debug = False
_chainedTo = None
def __init__(self, canceller=None):
"""
Initialize a L{Deferred}.
@param canceller: a callable used to stop the pending operation
scheduled by this L{Deferred} when L{Deferred.cancel} is
invoked. The canceller will be passed the deferred whose
cancelation is requested (i.e., self).
If a canceller is not given, or does not invoke its argument's
C{callback} or C{errback} method, L{Deferred.cancel} will
invoke L{Deferred.errback} with a L{CancelledError}.
Note that if a canceller is not given, C{callback} or
C{errback} may still be invoked exactly once, even though
defer.py will have already invoked C{errback}, as described
above. This allows clients of code which returns a L{Deferred}
to cancel it without requiring the L{Deferred} instantiator to
provide any specific implementation support for cancellation.
New in 10.1.
@type canceller: a 1-argument callable which takes a L{Deferred}. The
return result is ignored.
"""
self.callbacks = []
self._canceller = canceller
if self.debug:
self._debugInfo = DebugInfo()
self._debugInfo.creator = traceback.format_stack()[:-1]
def addCallbacks(self, callback, errback=None,
callbackArgs=None, callbackKeywords=None,
errbackArgs=None, errbackKeywords=None):
"""
Add a pair of callbacks (success and error) to this L{Deferred}.
These will be executed when the 'master' callback is run.
@return: C{self}.
@rtype: a L{Deferred}
"""
assert callable(callback)
assert errback is None or callable(errback)
cbs = ((callback, callbackArgs, callbackKeywords),
(errback or (passthru), errbackArgs, errbackKeywords))
self.callbacks.append(cbs)
if self.called:
self._runCallbacks()
return self
def addCallback(self, callback, *args, **kw):
"""
Convenience method for adding just a callback.
See L{addCallbacks}.
"""
return self.addCallbacks(callback, callbackArgs=args,
callbackKeywords=kw)
def addErrback(self, errback, *args, **kw):
"""
Convenience method for adding just an errback.
See L{addCallbacks}.
"""
return self.addCallbacks(passthru, errback,
errbackArgs=args,
errbackKeywords=kw)
def addBoth(self, callback, *args, **kw):
"""
Convenience method for adding a single callable as both a callback
and an errback.
See L{addCallbacks}.
"""
return self.addCallbacks(callback, callback,
callbackArgs=args, errbackArgs=args,
callbackKeywords=kw, errbackKeywords=kw)
def addTimeout(self, timeout, clock, onTimeoutCancel=None):
"""
Time out this L{Deferred} by scheduling it to be cancelled after
C{timeout} seconds.
The timeout encompasses all the callbacks and errbacks added to this
L{defer.Deferred} before the call to L{addTimeout}, and none added
after the call.
If this L{Deferred} gets timed out, it errbacks with a L{TimeoutError},
unless a cancelable function was passed to its initialization or unless
a different C{onTimeoutCancel} callable is provided.
@param timeout: number of seconds to wait before timing out this
L{Deferred}
@type timeout: L{int}
@param clock: The object which will be used to schedule the timeout.
@type clock: L{twisted.internet.interfaces.IReactorTime}
@param onTimeoutCancel: A callable which is called immediately after
this L{Deferred} times out, and not if this L{Deferred} is
otherwise cancelled before the timeout. It takes an arbitrary
value, which is the value of this L{Deferred} at that exact point
in time (probably a L{CancelledError} L{Failure}), and the
C{timeout}. The default callable (if none is provided) will
translate a L{CancelledError} L{Failure} into a L{TimeoutError}.
@type onTimeoutCancel: L{callable}
@return: C{self}.
@rtype: a L{Deferred}
@since: 16.5
"""
timedOut = [False]
def timeItOut():
timedOut[0] = True
self.cancel()
delayedCall = clock.callLater(timeout, timeItOut)
def convertCancelled(value):
# if C{deferred} was timed out, call the translation function,
# if provdied, otherwise just use L{cancelledToTimedOutError}
if timedOut[0]:
toCall = onTimeoutCancel or _cancelledToTimedOutError
return toCall(value, timeout)
return value
self.addBoth(convertCancelled)
def cancelTimeout(result):
# stop the pending call to cancel the deferred if it's been fired
if delayedCall.active():
delayedCall.cancel()
return result
self.addBoth(cancelTimeout)
return self
def chainDeferred(self, d):
"""
Chain another L{Deferred} to this L{Deferred}.
This method adds callbacks to this L{Deferred} to call C{d}'s callback
or errback, as appropriate. It is merely a shorthand way of performing
the following::
self.addCallbacks(d.callback, d.errback)
When you chain a deferred d2 to another deferred d1 with
d1.chainDeferred(d2), you are making d2 participate in the callback
chain of d1. Thus any event that fires d1 will also fire d2.
However, the converse is B{not} true; if d2 is fired d1 will not be
affected.
Note that unlike the case where chaining is caused by a L{Deferred}
being returned from a callback, it is possible to cause the call
stack size limit to be exceeded by chaining many L{Deferred}s
together with C{chainDeferred}.
@return: C{self}.
@rtype: a L{Deferred}
"""
d._chainedTo = self
return self.addCallbacks(d.callback, d.errback)
def callback(self, result):
"""
Run all success callbacks that have been added to this L{Deferred}.
Each callback will have its result passed as the first argument to
the next; this way, the callbacks act as a 'processing chain'. If
the success-callback returns a L{Failure} or raises an L{Exception},
processing will continue on the *error* callback chain. If a
callback (or errback) returns another L{Deferred}, this L{Deferred}
will be chained to it (and further callbacks will not run until that
L{Deferred} has a result).
An instance of L{Deferred} may only have either L{callback} or
L{errback} called on it, and only once.
@param result: The object which will be passed to the first callback
added to this L{Deferred} (via L{addCallback}).
@raise AlreadyCalledError: If L{callback} or L{errback} has already been
called on this L{Deferred}.
"""
assert not isinstance(result, Deferred)
self._startRunCallbacks(result)
def errback(self, fail=None):
"""
Run all error callbacks that have been added to this L{Deferred}.
Each callback will have its result passed as the first
argument to the next; this way, the callbacks act as a
'processing chain'. Also, if the error-callback returns a non-Failure
or doesn't raise an L{Exception}, processing will continue on the
*success*-callback chain.
If the argument that's passed to me is not a L{failure.Failure} instance,
it will be embedded in one. If no argument is passed, a
L{failure.Failure} instance will be created based on the current
traceback stack.
Passing a string as `fail' is deprecated, and will be punished with
a warning message.
An instance of L{Deferred} may only have either L{callback} or
L{errback} called on it, and only once.
@param fail: The L{Failure} object which will be passed to the first
errback added to this L{Deferred} (via L{addErrback}).
Alternatively, a L{Exception} instance from which a L{Failure} will
be constructed (with no traceback) or L{None} to create a L{Failure}
instance from the current exception state (with a traceback).
@raise AlreadyCalledError: If L{callback} or L{errback} has already been
called on this L{Deferred}.
@raise NoCurrentExceptionError: If C{fail} is L{None} but there is
no current exception state.
"""
if fail is None:
fail = failure.Failure(captureVars=self.debug)
elif not isinstance(fail, failure.Failure):
fail = failure.Failure(fail)
self._startRunCallbacks(fail)
def pause(self):
"""
Stop processing on a L{Deferred} until L{unpause}() is called.
"""
self.paused = self.paused + 1
def unpause(self):
"""
Process all callbacks made since L{pause}() was called.
"""
self.paused = self.paused - 1
if self.paused:
return
if self.called:
self._runCallbacks()
def cancel(self):
"""
Cancel this L{Deferred}.
If the L{Deferred} has not yet had its C{errback} or C{callback} method
invoked, call the canceller function provided to the constructor. If
that function does not invoke C{callback} or C{errback}, or if no
canceller function was provided, errback with L{CancelledError}.
If this L{Deferred} is waiting on another L{Deferred}, forward the
cancellation to the other L{Deferred}.
"""
if not self.called:
canceller = self._canceller
if canceller:
canceller(self)
else:
# Arrange to eat the callback that will eventually be fired
# since there was no real canceller.
self._suppressAlreadyCalled = True
if not self.called:
# There was no canceller, or the canceller didn't call
# callback or errback.
self.errback(failure.Failure(CancelledError()))
elif isinstance(self.result, Deferred):
# Waiting for another deferred -- cancel it instead.
self.result.cancel()
def _startRunCallbacks(self, result):
if self.called:
if self._suppressAlreadyCalled:
self._suppressAlreadyCalled = False
return
if self.debug:
if self._debugInfo is None:
self._debugInfo = DebugInfo()
extra = "\n" + self._debugInfo._getDebugTracebacks()
raise AlreadyCalledError(extra)
raise AlreadyCalledError
if self.debug:
if self._debugInfo is None:
self._debugInfo = DebugInfo()
self._debugInfo.invoker = traceback.format_stack()[:-2]
self.called = True
self.result = result
self._runCallbacks()
def _continuation(self):
"""
Build a tuple of callback and errback with L{_CONTINUE}.
"""
return ((_CONTINUE, (self,), None),
(_CONTINUE, (self,), None))
def _runCallbacks(self):
"""
Run the chain of callbacks once a result is available.
This consists of a simple loop over all of the callbacks, calling each
with the current result and making the current result equal to the
return value (or raised exception) of that call.
If L{_runningCallbacks} is true, this loop won't run at all, since
it is already running above us on the call stack. If C{self.paused} is
true, the loop also won't run, because that's what it means to be
paused.
The loop will terminate before processing all of the callbacks if a
L{Deferred} without a result is encountered.
If a L{Deferred} I{with} a result is encountered, that result is taken
and the loop proceeds.
@note: The implementation is complicated slightly by the fact that
chaining (associating two L{Deferred}s with each other such that one
will wait for the result of the other, as happens when a Deferred is
returned from a callback on another L{Deferred}) is supported
iteratively rather than recursively, to avoid running out of stack
frames when processing long chains.
"""
if self._runningCallbacks:
# Don't recursively run callbacks
return
# Keep track of all the Deferreds encountered while propagating results
# up a chain. The way a Deferred gets onto this stack is by having
# added its _continuation() to the callbacks list of a second Deferred
# and then that second Deferred being fired. ie, if ever had _chainedTo
# set to something other than None, you might end up on this stack.
chain = [self]
while chain:
current = chain[-1]
if current.paused:
# This Deferred isn't going to produce a result at all. All the
# Deferreds up the chain waiting on it will just have to...
# wait.
return
finished = True
current._chainedTo = None
while current.callbacks:
item = current.callbacks.pop(0)
callback, args, kw = item[
isinstance(current.result, failure.Failure)]
args = args or ()
kw = kw or {}
# Avoid recursion if we can.
if callback is _CONTINUE:
# Give the waiting Deferred our current result and then
# forget about that result ourselves.
chainee = args[0]
chainee.result = current.result
current.result = None
# Making sure to update _debugInfo
if current._debugInfo is not None:
current._debugInfo.failResult = None
chainee.paused -= 1
chain.append(chainee)
# Delay cleaning this Deferred and popping it from the chain
# until after we've dealt with chainee.
finished = False
break
try:
current._runningCallbacks = True
try:
current.result = callback(current.result, *args, **kw)
if current.result is current:
warnAboutFunction(
callback,
"Callback returned the Deferred "
"it was attached to; this breaks the "
"callback chain and will raise an "
"exception in the future.")
finally:
current._runningCallbacks = False
except:
# Including full frame information in the Failure is quite
# expensive, so we avoid it unless self.debug is set.
current.result = failure.Failure(captureVars=self.debug)
else:
if isinstance(current.result, Deferred):
# The result is another Deferred. If it has a result,
# we can take it and keep going.
resultResult = getattr(current.result, 'result', _NO_RESULT)
if resultResult is _NO_RESULT or isinstance(resultResult, Deferred) or current.result.paused:
# Nope, it didn't. Pause and chain.
current.pause()
current._chainedTo = current.result
# Note: current.result has no result, so it's not
# running its callbacks right now. Therefore we can
# append to the callbacks list directly instead of
# using addCallbacks.
current.result.callbacks.append(current._continuation())
break
else:
# Yep, it did. Steal it.
current.result.result = None
# Make sure _debugInfo's failure state is updated.
if current.result._debugInfo is not None:
current.result._debugInfo.failResult = None
current.result = resultResult
if finished:
# As much of the callback chain - perhaps all of it - as can be
# processed right now has been. The current Deferred is waiting on
# another Deferred or for more callbacks. Before finishing with it,
# make sure its _debugInfo is in the proper state.
if isinstance(current.result, failure.Failure):
# Stash the Failure in the _debugInfo for unhandled error
# reporting.
current.result.cleanFailure()
if current._debugInfo is None:
current._debugInfo = DebugInfo()
current._debugInfo.failResult = current.result
else:
# Clear out any Failure in the _debugInfo, since the result
# is no longer a Failure.
if current._debugInfo is not None:
current._debugInfo.failResult = None
# This Deferred is done, pop it from the chain and move back up
# to the Deferred which supplied us with our result.
chain.pop()
def __str__(self):
"""
Return a string representation of this C{Deferred}.
"""
cname = self.__class__.__name__
result = getattr(self, 'result', _NO_RESULT)
myID = id(self)
if self._chainedTo is not None:
result = ' waiting on Deferred at 0x%x' % (id(self._chainedTo),)
elif result is _NO_RESULT:
result = ''
else:
result = ' current result: %r' % (result,)
return "<%s at 0x%x%s>" % (cname, myID, result)
__repr__ = __str__
def __iter__(self):
return self
@failure._extraneous
def send(self, value=None):
if self.paused:
# If we're paused, we have no result to give
return self
result = getattr(self, 'result', _NO_RESULT)
if result is _NO_RESULT:
return self
if isinstance(result, failure.Failure):
# Clear the failure on debugInfo so it doesn't raise "unhandled
# exception"
self._debugInfo.failResult = None
result.value.__failure__ = result
raise result.value
else:
raise StopIteration(result)
# For PEP-492 support (async/await)
__await__ = __iter__
__next__ = send
def asFuture(self, loop):
"""
Adapt a L{Deferred} into a L{asyncio.Future} which is bound to C{loop}.
@note: converting a L{Deferred} to an L{asyncio.Future} consumes both
its result and its errors, so this method implicitly converts
C{self} into a L{Deferred} firing with L{None}, regardless of what
its result previously would have been.
@since: Twisted 17.5.0
@param loop: The asyncio event loop to bind the L{asyncio.Future} to.
@type loop: L{asyncio.AbstractEventLoop} or similar
@param deferred: The Deferred to adapt.
@type deferred: L{Deferred}
@return: A Future which will fire when the Deferred fires.
@rtype: L{asyncio.Future}
"""
try:
createFuture = loop.create_future
except AttributeError:
from asyncio import Future
def createFuture():
return Future(loop=loop)
future = createFuture()
def checkCancel(futureAgain):
if futureAgain.cancelled():
self.cancel()
def maybeFail(failure):
if not future.cancelled():
future.set_exception(failure.value)
def maybeSucceed(result):
if not future.cancelled():
future.set_result(result)
self.addCallbacks(maybeSucceed, maybeFail)
future.add_done_callback(checkCancel)
return future
@classmethod
def fromFuture(cls, future):
"""
Adapt an L{asyncio.Future} to a L{Deferred}.
@note: This creates a L{Deferred} from a L{asyncio.Future}, I{not} from
a C{coroutine}; in other words, you will need to call
L{asyncio.ensure_future},
L{asyncio.loop.create_task} or create an
L{asyncio.Task} yourself to get from a C{coroutine} to a
L{asyncio.Future} if what you have is an awaitable coroutine and
not a L{asyncio.Future}. (The length of this list of techniques is
exactly why we have left it to the caller!)
@since: Twisted 17.5.0
@param future: The Future to adapt.
@type future: L{asyncio.Future}
@return: A Deferred which will fire when the Future fires.
@rtype: L{Deferred}
"""
def adapt(result):
try:
extracted = result.result()
except:
extracted = failure.Failure()
adapt.actual.callback(extracted)
futureCancel = object()
def cancel(reself):
future.cancel()
reself.callback(futureCancel)
self = cls(cancel)
adapt.actual = self
def uncancel(result):
if result is futureCancel:
adapt.actual = Deferred()
return adapt.actual
return result
self.addCallback(uncancel)
future.add_done_callback(adapt)
return self
def _cancelledToTimedOutError(value, timeout):
"""
A default translation function that translates L{Failure}s that are
L{CancelledError}s to L{TimeoutError}s.
@param value: Anything
@type value: Anything
@param timeout: The timeout
@type timeout: L{int}
@rtype: C{value}
@raise: L{TimeoutError}
@since: 16.5
"""
if isinstance(value, failure.Failure):
value.trap(CancelledError)
raise TimeoutError(timeout, "Deferred")
return value
def ensureDeferred(coro):
"""
Schedule the execution of a coroutine that awaits/yields from L{Deferred}s,
wrapping it in a L{Deferred} that will fire on success/failure of the
coroutine. If a Deferred is passed to this function, it will be returned
directly (mimicing C{asyncio}'s C{ensure_future} function).
Coroutine functions return a coroutine object, similar to how generators
work. This function turns that coroutine into a Deferred, meaning that it
can be used in regular Twisted code. For example::
import treq
from twisted.internet.defer import ensureDeferred
from twisted.internet.task import react
async def crawl(pages):
results = {}
for page in pages:
results[page] = await treq.content(await treq.get(page))
return results
def main(reactor):
pages = [
"http://localhost:8080"
]
d = ensureDeferred(crawl(pages))
d.addCallback(print)
return d
react(main)
@param coro: The coroutine object to schedule, or a L{Deferred}.
@type coro: A Python 3.5+ C{async def} C{coroutine}, a Python 3.4+
C{yield from} using L{types.GeneratorType}, or a L{Deferred}.
@rtype: L{Deferred}
"""
from types import GeneratorType
if version_info >= (3, 4, 0):
from asyncio import iscoroutine
if iscoroutine(coro) or isinstance(coro, GeneratorType):
return _cancellableInlineCallbacks(coro)
if not isinstance(coro, Deferred):
raise ValueError("%r is not a coroutine or a Deferred" % (coro,))
# Must be a Deferred
return coro
@_oldStyle
class DebugInfo:
"""
Deferred debug helper.
"""
failResult = None
def _getDebugTracebacks(self):
info = ''
if hasattr(self, "creator"):
info += " C: Deferred was created:\n C:"
info += "".join(self.creator).rstrip().replace("\n", "\n C:")
info += "\n"
if hasattr(self, "invoker"):
info += " I: First Invoker was:\n I:"
info += "".join(self.invoker).rstrip().replace("\n", "\n I:")
info += "\n"
return info
def __del__(self):
"""
Print tracebacks and die.
If the *last* (and I do mean *last*) callback leaves me in an error
state, print a traceback (if said errback is a L{Failure}).
"""
if self.failResult is not None:
# Note: this is two separate messages for compatibility with
# earlier tests; arguably it should be a single error message.
log.critical("Unhandled error in Deferred:",
isError=True)
debugInfo = self._getDebugTracebacks()
if debugInfo:
format = "(debug: {debugInfo})"
else:
format = None
log.failure(format,
self.failResult,
debugInfo=debugInfo)
@comparable
class FirstError(Exception):
"""
First error to occur in a L{DeferredList} if C{fireOnOneErrback} is set.
@ivar subFailure: The L{Failure} that occurred.
@type subFailure: L{Failure}
@ivar index: The index of the L{Deferred} in the L{DeferredList} where
it happened.
@type index: L{int}
"""
def __init__(self, failure, index):
Exception.__init__(self, failure, index)
self.subFailure = failure
self.index = index
def __repr__(self):
"""
The I{repr} of L{FirstError} instances includes the repr of the
wrapped failure's exception and the index of the L{FirstError}.
"""
return 'FirstError[#%d, %r]' % (self.index, self.subFailure.value)
def __str__(self):
"""
The I{str} of L{FirstError} instances includes the I{str} of the
entire wrapped failure (including its traceback and exception) and
the index of the L{FirstError}.
"""
return 'FirstError[#%d, %s]' % (self.index, self.subFailure)