Coverage for /pythoncovmergedfiles/medio/medio/usr/lib/python3.9/asyncio/base_futures.py: 26%

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