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