• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1__all__ = ()
2
3import reprlib
4from _thread import get_ident
5
6from . import format_helpers
7
8# States for Future.
9_PENDING = 'PENDING'
10_CANCELLED = 'CANCELLED'
11_FINISHED = 'FINISHED'
12
13
14def 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
25def _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# bpo-42183: _repr_running is needed for repr protection
46# when a Future or Task result contains itself directly or indirectly.
47# The logic is borrowed from @reprlib.recursive_repr decorator.
48# Unfortunately, the direct decorator usage is impossible because of
49# AttributeError: '_asyncio.Task' object has no attribute '__module__' error.
50#
51# After fixing this thing we can return to the decorator based approach.
52_repr_running = set()
53
54
55def _future_repr_info(future):
56    # (Future) -> str
57    """helper function for Future.__repr__"""
58    info = [future._state.lower()]
59    if future._state == _FINISHED:
60        if future._exception is not None:
61            info.append(f'exception={future._exception!r}')
62        else:
63            key = id(future), get_ident()
64            if key in _repr_running:
65                result = '...'
66            else:
67                _repr_running.add(key)
68                try:
69                    # use reprlib to limit the length of the output, especially
70                    # for very long strings
71                    result = reprlib.repr(future._result)
72                finally:
73                    _repr_running.discard(key)
74            info.append(f'result={result}')
75    if future._callbacks:
76        info.append(_format_callbacks(future._callbacks))
77    if future._source_traceback:
78        frame = future._source_traceback[-1]
79        info.append(f'created at {frame[0]}:{frame[1]}')
80    return info
81