File Explorer

1__all__ = ()2 3import reprlib4from _thread import get_ident5 6from . import format_helpers7 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 advertising18    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') and22            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 protection46# 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 of49# 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) -> str57    """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, especially70                    # for very long strings71                    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 info81