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