python (3.12.0)
1 __all__ = 'iscoroutinefunction', 'iscoroutine'
2
3 import collections.abc
4 import inspect
5 import os
6 import sys
7 import types
8
9
10 def _is_debug_mode():
11 # See: https://docs.python.org/3/library/asyncio-dev.html#asyncio-debug-mode.
12 return sys.flags.dev_mode or (not sys.flags.ignore_environment and
13 bool(os.environ.get('PYTHONASYNCIODEBUG')))
14
15
16 # A marker for iscoroutinefunction.
17 _is_coroutine = object()
18
19
20 def iscoroutinefunction(func):
21 """Return True if func is a decorated coroutine function."""
22 return (inspect.iscoroutinefunction(func) or
23 getattr(func, '_is_coroutine', None) is _is_coroutine)
24
25
26 # Prioritize native coroutine check to speed-up
27 # asyncio.iscoroutine.
28 _COROUTINE_TYPES = (types.CoroutineType, collections.abc.Coroutine)
29 _iscoroutine_typecache = set()
30
31
32 def iscoroutine(obj):
33 """Return True if obj is a coroutine object."""
34 if type(obj) in _iscoroutine_typecache:
35 return True
36
37 if isinstance(obj, _COROUTINE_TYPES):
38 # Just in case we don't want to cache more than 100
39 # positive types. That shouldn't ever happen, unless
40 # someone stressing the system on purpose.
41 if len(_iscoroutine_typecache) < 100:
42 _iscoroutine_typecache.add(type(obj))
43 return True
44 else:
45 return False
46
47
48 def _format_coroutine(coro):
49 assert iscoroutine(coro)
50
51 def get_name(coro):
52 # Coroutines compiled with Cython sometimes don't have
53 # proper __qualname__ or __name__. While that is a bug
54 # in Cython, asyncio shouldn't crash with an AttributeError
55 # in its __repr__ functions.
56 if hasattr(coro, '__qualname__') and coro.__qualname__:
57 coro_name = coro.__qualname__
58 elif hasattr(coro, '__name__') and coro.__name__:
59 coro_name = coro.__name__
60 else:
61 # Stop masking Cython bugs, expose them in a friendly way.
62 coro_name = f'<{type(coro).__name__} without __name__>'
63 return f'{coro_name}()'
64
65 def is_running(coro):
66 try:
67 return coro.cr_running
68 except AttributeError:
69 try:
70 return coro.gi_running
71 except AttributeError:
72 return False
73
74 coro_code = None
75 if hasattr(coro, 'cr_code') and coro.cr_code:
76 coro_code = coro.cr_code
77 elif hasattr(coro, 'gi_code') and coro.gi_code:
78 coro_code = coro.gi_code
79
80 coro_name = get_name(coro)
81
82 if not coro_code:
83 # Built-in types might not have __qualname__ or __name__.
84 if is_running(coro):
85 return f'{coro_name} running'
86 else:
87 return coro_name
88
89 coro_frame = None
90 if hasattr(coro, 'gi_frame') and coro.gi_frame:
91 coro_frame = coro.gi_frame
92 elif hasattr(coro, 'cr_frame') and coro.cr_frame:
93 coro_frame = coro.cr_frame
94
95 # If Cython's coroutine has a fake code object without proper
96 # co_filename -- expose that.
97 filename = coro_code.co_filename or '<empty co_filename>'
98
99 lineno = 0
100
101 if coro_frame is not None:
102 lineno = coro_frame.f_lineno
103 coro_repr = f'{coro_name} running at {filename}:{lineno}'
104
105 else:
106 lineno = coro_code.co_firstlineno
107 coro_repr = f'{coro_name} done, defined at {filename}:{lineno}'
108
109 return coro_repr
Browse
Usage
Versions