• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1__all__ = 'iscoroutinefunction', 'iscoroutine'
2
3import collections.abc
4import inspect
5import os
6import sys
7import types
8
9
10def _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
20def 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
32def 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
48def _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
110