python (3.11.7)
1 __all__ = ()
2
3 import reprlib
4 from _thread import get_ident
5
6 from . import format_helpers
7
8 # States for Future.
9 _PENDING = 'PENDING'
10 _CANCELLED = 'CANCELLED'
11 _FINISHED = 'FINISHED'
12
13
14 def isfuture(obj):
15 """Check for a Future.
16
17 This returns True when obj is a Future instance or is advertising
18 itself as duck-type compatible by setting _asyncio_future_blocking.
19 See comment in Future for more details.
20 """
21 return (hasattr(obj.__class__, '_asyncio_future_blocking') and
22 obj._asyncio_future_blocking is not None)
23
24
25 def _format_callbacks(cb):
26 """helper function for Future.__repr__"""
27 size = len(cb)
28 if not size:
29 cb = ''
30
31 def format_cb(callback):
32 return format_helpers._format_callback_source(callback, ())
33
34 if size == 1:
35 cb = format_cb(cb[0][0])
36 elif size == 2:
37 cb = '{}, {}'.format(format_cb(cb[0][0]), format_cb(cb[1][0]))
38 elif size > 2:
39 cb = '{}, <{} more>, {}'.format(format_cb(cb[0][0]),
40 size - 2,
41 format_cb(cb[-1][0]))
42 return f'cb=[{cb}]'
43
44
45 def _future_repr_info(future):
46 # (Future) -> str
47 """helper function for Future.__repr__"""
48 info = [future._state.lower()]
49 if future._state == _FINISHED:
50 if future._exception is not None:
51 info.append(f'exception={future._exception!r}')
52 else:
53 # use reprlib to limit the length of the output, especially
54 # for very long strings
55 result = reprlib.repr(future._result)
56 info.append(f'result={result}')
57 if future._callbacks:
58 info.append(_format_callbacks(future._callbacks))
59 if future._source_traceback:
60 frame = future._source_traceback[-1]
61 info.append(f'created at {frame[0]}:{frame[1]}')
62 return info
63
64
65 @reprlib.recursive_repr()
66 def _future_repr(future):
67 info = ' '.join(_future_repr_info(future))
68 return f'<{future.__class__.__name__} {info}>'
Browse
Usage
Versions