aboutsummaryrefslogtreecommitdiffstats
path: root/contrib/tools/python3/src/Lib/asyncio/coroutines.py
diff options
context:
space:
mode:
authormonster <monster@ydb.tech>2022-07-07 14:41:37 +0300
committermonster <monster@ydb.tech>2022-07-07 14:41:37 +0300
commit06e5c21a835c0e923506c4ff27929f34e00761c2 (patch)
tree75efcbc6854ef9bd476eb8bf00cc5c900da436a2 /contrib/tools/python3/src/Lib/asyncio/coroutines.py
parent03f024c4412e3aa613bb543cf1660176320ba8f4 (diff)
downloadydb-06e5c21a835c0e923506c4ff27929f34e00761c2.tar.gz
fix ya.make
Diffstat (limited to 'contrib/tools/python3/src/Lib/asyncio/coroutines.py')
-rw-r--r--contrib/tools/python3/src/Lib/asyncio/coroutines.py269
1 files changed, 0 insertions, 269 deletions
diff --git a/contrib/tools/python3/src/Lib/asyncio/coroutines.py b/contrib/tools/python3/src/Lib/asyncio/coroutines.py
deleted file mode 100644
index 9664ea74d75..00000000000
--- a/contrib/tools/python3/src/Lib/asyncio/coroutines.py
+++ /dev/null
@@ -1,269 +0,0 @@
-__all__ = 'coroutine', 'iscoroutinefunction', 'iscoroutine'
-
-import collections.abc
-import functools
-import inspect
-import os
-import sys
-import traceback
-import types
-import warnings
-
-from . import base_futures
-from . import constants
-from . import format_helpers
-from .log import logger
-
-
-def _is_debug_mode():
- # If you set _DEBUG to true, @coroutine will wrap the resulting
- # generator objects in a CoroWrapper instance (defined below). That
- # instance will log a message when the generator is never iterated
- # over, which may happen when you forget to use "await" or "yield from"
- # with a coroutine call.
- # Note that the value of the _DEBUG flag is taken
- # when the decorator is used, so to be of any use it must be set
- # before you define your coroutines. A downside of using this feature
- # is that tracebacks show entries for the CoroWrapper.__next__ method
- # when _DEBUG is true.
- return sys.flags.dev_mode or (not sys.flags.ignore_environment and
- bool(os.environ.get('PYTHONASYNCIODEBUG')))
-
-
-_DEBUG = _is_debug_mode()
-
-
-class CoroWrapper:
- # Wrapper for coroutine object in _DEBUG mode.
-
- def __init__(self, gen, func=None):
- assert inspect.isgenerator(gen) or inspect.iscoroutine(gen), gen
- self.gen = gen
- self.func = func # Used to unwrap @coroutine decorator
- self._source_traceback = format_helpers.extract_stack(sys._getframe(1))
- self.__name__ = getattr(gen, '__name__', None)
- self.__qualname__ = getattr(gen, '__qualname__', None)
-
- def __repr__(self):
- coro_repr = _format_coroutine(self)
- if self._source_traceback:
- frame = self._source_traceback[-1]
- coro_repr += f', created at {frame[0]}:{frame[1]}'
-
- return f'<{self.__class__.__name__} {coro_repr}>'
-
- def __iter__(self):
- return self
-
- def __next__(self):
- return self.gen.send(None)
-
- def send(self, value):
- return self.gen.send(value)
-
- def throw(self, type, value=None, traceback=None):
- return self.gen.throw(type, value, traceback)
-
- def close(self):
- return self.gen.close()
-
- @property
- def gi_frame(self):
- return self.gen.gi_frame
-
- @property
- def gi_running(self):
- return self.gen.gi_running
-
- @property
- def gi_code(self):
- return self.gen.gi_code
-
- def __await__(self):
- return self
-
- @property
- def gi_yieldfrom(self):
- return self.gen.gi_yieldfrom
-
- def __del__(self):
- # Be careful accessing self.gen.frame -- self.gen might not exist.
- gen = getattr(self, 'gen', None)
- frame = getattr(gen, 'gi_frame', None)
- if frame is not None and frame.f_lasti == -1:
- msg = f'{self!r} was never yielded from'
- tb = getattr(self, '_source_traceback', ())
- if tb:
- tb = ''.join(traceback.format_list(tb))
- msg += (f'\nCoroutine object created at '
- f'(most recent call last, truncated to '
- f'{constants.DEBUG_STACK_DEPTH} last lines):\n')
- msg += tb.rstrip()
- logger.error(msg)
-
-
-def coroutine(func):
- """Decorator to mark coroutines.
-
- If the coroutine is not yielded from before it is destroyed,
- an error message is logged.
- """
- warnings.warn('"@coroutine" decorator is deprecated since Python 3.8, use "async def" instead',
- DeprecationWarning,
- stacklevel=2)
- if inspect.iscoroutinefunction(func):
- # In Python 3.5 that's all we need to do for coroutines
- # defined with "async def".
- return func
-
- if inspect.isgeneratorfunction(func):
- coro = func
- else:
- @functools.wraps(func)
- def coro(*args, **kw):
- res = func(*args, **kw)
- if (base_futures.isfuture(res) or inspect.isgenerator(res) or
- isinstance(res, CoroWrapper)):
- res = yield from res
- else:
- # If 'res' is an awaitable, run it.
- try:
- await_meth = res.__await__
- except AttributeError:
- pass
- else:
- if isinstance(res, collections.abc.Awaitable):
- res = yield from await_meth()
- return res
-
- coro = types.coroutine(coro)
- if not _DEBUG:
- wrapper = coro
- else:
- @functools.wraps(func)
- def wrapper(*args, **kwds):
- w = CoroWrapper(coro(*args, **kwds), func=func)
- if w._source_traceback:
- del w._source_traceback[-1]
- # Python < 3.5 does not implement __qualname__
- # on generator objects, so we set it manually.
- # We use getattr as some callables (such as
- # functools.partial may lack __qualname__).
- w.__name__ = getattr(func, '__name__', None)
- w.__qualname__ = getattr(func, '__qualname__', None)
- return w
-
- wrapper._is_coroutine = _is_coroutine # For iscoroutinefunction().
- return wrapper
-
-
-# A marker for iscoroutinefunction.
-_is_coroutine = object()
-
-
-def iscoroutinefunction(func):
- """Return True if func is a decorated coroutine function."""
- return (inspect.iscoroutinefunction(func) or
- getattr(func, '_is_coroutine', None) is _is_coroutine)
-
-
-# Prioritize native coroutine check to speed-up
-# asyncio.iscoroutine.
-_COROUTINE_TYPES = (types.CoroutineType, types.GeneratorType,
- collections.abc.Coroutine, CoroWrapper)
-_iscoroutine_typecache = set()
-
-
-def iscoroutine(obj):
- """Return True if obj is a coroutine object."""
- if type(obj) in _iscoroutine_typecache:
- return True
-
- if isinstance(obj, _COROUTINE_TYPES):
- # Just in case we don't want to cache more than 100
- # positive types. That shouldn't ever happen, unless
- # someone stressing the system on purpose.
- if len(_iscoroutine_typecache) < 100:
- _iscoroutine_typecache.add(type(obj))
- return True
- else:
- return False
-
-
-def _format_coroutine(coro):
- assert iscoroutine(coro)
-
- is_corowrapper = isinstance(coro, CoroWrapper)
-
- def get_name(coro):
- # Coroutines compiled with Cython sometimes don't have
- # proper __qualname__ or __name__. While that is a bug
- # in Cython, asyncio shouldn't crash with an AttributeError
- # in its __repr__ functions.
- if is_corowrapper:
- return format_helpers._format_callback(coro.func, (), {})
-
- if hasattr(coro, '__qualname__') and coro.__qualname__:
- coro_name = coro.__qualname__
- elif hasattr(coro, '__name__') and coro.__name__:
- coro_name = coro.__name__
- else:
- # Stop masking Cython bugs, expose them in a friendly way.
- coro_name = f'<{type(coro).__name__} without __name__>'
- return f'{coro_name}()'
-
- def is_running(coro):
- try:
- return coro.cr_running
- except AttributeError:
- try:
- return coro.gi_running
- except AttributeError:
- return False
-
- coro_code = None
- if hasattr(coro, 'cr_code') and coro.cr_code:
- coro_code = coro.cr_code
- elif hasattr(coro, 'gi_code') and coro.gi_code:
- coro_code = coro.gi_code
-
- coro_name = get_name(coro)
-
- if not coro_code:
- # Built-in types might not have __qualname__ or __name__.
- if is_running(coro):
- return f'{coro_name} running'
- else:
- return coro_name
-
- coro_frame = None
- if hasattr(coro, 'gi_frame') and coro.gi_frame:
- coro_frame = coro.gi_frame
- elif hasattr(coro, 'cr_frame') and coro.cr_frame:
- coro_frame = coro.cr_frame
-
- # If Cython's coroutine has a fake code object without proper
- # co_filename -- expose that.
- filename = coro_code.co_filename or '<empty co_filename>'
-
- lineno = 0
- if (is_corowrapper and
- coro.func is not None and
- not inspect.isgeneratorfunction(coro.func)):
- source = format_helpers._get_function_source(coro.func)
- if source is not None:
- filename, lineno = source
- if coro_frame is None:
- coro_repr = f'{coro_name} done, defined at {filename}:{lineno}'
- else:
- coro_repr = f'{coro_name} running, defined at {filename}:{lineno}'
-
- elif coro_frame is not None:
- lineno = coro_frame.f_lineno
- coro_repr = f'{coro_name} running at {filename}:{lineno}'
-
- else:
- lineno = coro_code.co_firstlineno
- coro_repr = f'{coro_name} done, defined at {filename}:{lineno}'
-
- return coro_repr