+++ /dev/null
-__all__ = ['coroutine',
- 'iscoroutinefunction', 'iscoroutine',
- 'From', 'Return']
-
-import functools
-import inspect
-import opcode
-import os
-import sys
-import textwrap
-import traceback
-import types
-
-from . import compat
-from . import events
-from . import futures
-from .log import logger
-
-
-# Opcode of "yield from" instruction
-_YIELD_FROM = opcode.opmap.get('YIELD_FROM', None)
-
-# 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 "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.
-_DEBUG = bool(os.environ.get('TROLLIUSDEBUG'))
-
-
-try:
- _types_coroutine = types.coroutine
-except AttributeError:
- _types_coroutine = None
-
-try:
- _inspect_iscoroutinefunction = inspect.iscoroutinefunction
-except AttributeError:
- _inspect_iscoroutinefunction = lambda func: False
-
-try:
- from collections.abc import Coroutine as _CoroutineABC, \
- Awaitable as _AwaitableABC
-except ImportError:
- _CoroutineABC = _AwaitableABC = None
-
-
-if _YIELD_FROM is not None:
- # Check for CPython issue #21209
- exec('''if 1:
- def has_yield_from_bug():
- class MyGen:
- def __init__(self):
- self.send_args = None
- def __iter__(self):
- return self
- def __next__(self):
- return 42
- def send(self, *what):
- self.send_args = what
- return None
- def yield_from_gen(gen):
- yield from gen
- value = (1, 2, 3)
- gen = MyGen()
- coro = yield_from_gen(gen)
- next(coro)
- coro.send(value)
- return gen.send_args != (value,)
-''')
- _YIELD_FROM_BUG = has_yield_from_bug()
- del has_yield_from_bug
-else:
- _YIELD_FROM_BUG = False
-
-
-if compat.PY35:
- return_base_class = Exception
-else:
- return_base_class = StopIteration
-
-class ReturnException(return_base_class):
- def __init__(self, *args):
- return_base_class.__init__(self)
- if not args:
- self.value = None
- elif len(args) == 1:
- self.value = args[0]
- else:
- self.value = args
- self.raised = False
- if _DEBUG:
- frame = sys._getframe(1)
- self._source_traceback = traceback.extract_stack(frame)
- # explicitly clear the reference to avoid reference cycles
- frame = None
- else:
- self._source_traceback = None
-
- def __del__(self):
- if self.raised:
- return
-
- fmt = 'Return(%r) used without raise'
- if self._source_traceback:
- fmt += '\nReturn created at (most recent call last):\n'
- tb = ''.join(traceback.format_list(self._source_traceback))
- fmt += tb.rstrip()
- logger.error(fmt, self.value)
-
-
-if compat.PY33 and not compat.PY35:
- # Don't use the Return class on Python 3.3 and 3.4 to support asyncio
- # coroutines (to avoid the warning emited in Return destructor).
- #
- # The problem is that ReturnException inherits from StopIteration.
- # "yield from trollius_coroutine". Task._step() does not receive the Return
- # exception, because "yield from" handles it internally. So it's not
- # possible to set the raised attribute to True to avoid the warning in
- # Return destructor.
- def Return(*args):
- if not args:
- value = None
- elif len(args) == 1:
- value = args[0]
- else:
- value = args
- return StopIteration(value)
-else:
- Return = ReturnException
-
-
-def debug_wrapper(gen):
- # This function is called from 'sys.set_coroutine_wrapper'.
- # We only wrap here coroutines defined via 'async def' syntax.
- # Generator-based coroutines are wrapped in @coroutine
- # decorator.
- return CoroWrapper(gen, None)
-
-
-def _coroutine_at_yield_from(coro):
- """Test if the last instruction of a coroutine is "yield from".
-
- Return False if the coroutine completed.
- """
- frame = coro.gi_frame
- if frame is None:
- return False
- code = coro.gi_code
- assert frame.f_lasti >= 0
- offset = frame.f_lasti + 1
- instr = code.co_code[offset]
- return (instr == _YIELD_FROM)
-
-
-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 = traceback.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 += ', created at %s:%s' % (frame[0], frame[1])
- return '<%s %s>' % (self.__class__.__name__, coro_repr)
-
- def __iter__(self):
- return self
-
- def __next__(self):
- return next(self.gen)
- next = __next__
-
- if _YIELD_FROM_BUG:
- # For for CPython issue #21209: using "yield from" and a custom
- # generator, generator.send(tuple) unpacks the tuple instead of passing
- # the tuple unchanged. Check if the caller is a generator using "yield
- # from" to decide if the parameter should be unpacked or not.
- def send(self, *value):
- frame = sys._getframe()
- caller = frame.f_back
- assert caller.f_lasti >= 0
- if caller.f_code.co_code[caller.f_lasti] != _YIELD_FROM:
- value = value[0]
- return self.gen.send(value)
- else:
- def send(self, value):
- return self.gen.send(value)
-
- def throw(self, exc_type, exc_value=None, exc_tb=None):
- return self.gen.throw(exc_type, exc_value, exc_tb)
-
- 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
-
- if compat.PY35:
-
- __await__ = __iter__ # make compatible with 'await' expression
-
- @property
- def gi_yieldfrom(self):
- return self.gen.gi_yieldfrom
-
- @property
- def cr_await(self):
- return self.gen.cr_await
-
- @property
- def cr_running(self):
- return self.gen.cr_running
-
- @property
- def cr_code(self):
- return self.gen.cr_code
-
- @property
- def cr_frame(self):
- return self.gen.cr_frame
-
- 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 None:
- frame = getattr(gen, 'cr_frame', None)
- if frame is not None and frame.f_lasti == -1:
- msg = '%r was never yielded from' % self
- tb = getattr(self, '_source_traceback', ())
- if tb:
- tb = ''.join(traceback.format_list(tb))
- msg += ('\nCoroutine object created at '
- '(most recent call last):\n')
- msg += tb.rstrip()
- logger.error(msg)
-
-if not compat.PY34:
- # Backport functools.update_wrapper() from Python 3.4:
- # - Python 2.7 fails if assigned attributes don't exist
- # - Python 2.7 and 3.1 don't set the __wrapped__ attribute
- # - Python 3.2 and 3.3 set __wrapped__ before updating __dict__
- def _update_wrapper(wrapper,
- wrapped,
- assigned = functools.WRAPPER_ASSIGNMENTS,
- updated = functools.WRAPPER_UPDATES):
- """Update a wrapper function to look like the wrapped function
-
- wrapper is the function to be updated
- wrapped is the original function
- assigned is a tuple naming the attributes assigned directly
- from the wrapped function to the wrapper function (defaults to
- functools.WRAPPER_ASSIGNMENTS)
- updated is a tuple naming the attributes of the wrapper that
- are updated with the corresponding attribute from the wrapped
- function (defaults to functools.WRAPPER_UPDATES)
- """
- for attr in assigned:
- try:
- value = getattr(wrapped, attr)
- except AttributeError:
- pass
- else:
- setattr(wrapper, attr, value)
- for attr in updated:
- getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
- # Issue #17482: set __wrapped__ last so we don't inadvertently copy it
- # from the wrapped function when updating __dict__
- wrapper.__wrapped__ = wrapped
- # Return the wrapper so this can be used as a decorator via partial()
- return wrapper
-
- def _wraps(wrapped,
- assigned = functools.WRAPPER_ASSIGNMENTS,
- updated = functools.WRAPPER_UPDATES):
- """Decorator factory to apply update_wrapper() to a wrapper function
-
- Returns a decorator that invokes update_wrapper() with the decorated
- function as the wrapper argument and the arguments to wraps() as the
- remaining arguments. Default arguments are as for update_wrapper().
- This is a convenience function to simplify applying partial() to
- update_wrapper().
- """
- return functools.partial(_update_wrapper, wrapped=wrapped,
- assigned=assigned, updated=updated)
-else:
- _wraps = functools.wraps
-
-_PEP479 = (sys.version_info >= (3, 5))
-if _PEP479:
- # Need exec() because yield+return raises a SyntaxError on Python 2
- exec(textwrap.dedent('''
- def pep479_wrapper(func, coro_func):
- @_wraps(func)
- def pep479_wrapped(*args, **kw):
- coro = coro_func(*args, **kw)
- value = None
- error = None
- while True:
- try:
- if error is not None:
- value = coro.throw(error)
- elif value is not None:
- value = coro.send(value)
- else:
- value = next(coro)
- except RuntimeError:
- # FIXME: special case for
- # FIXME: "isinstance(exc.__context__, StopIteration)"?
- raise
- except StopIteration as exc:
- return exc.value
- except Return as exc:
- exc.raised = True
- return exc.value
- except BaseException as exc:
- raise
-
- try:
- value = yield value
- error = None
- except BaseException as exc:
- value = None
- error = exc
-
- return pep479_wrapped
- '''))
-
-
-def coroutine(func):
- """Decorator to mark coroutines.
-
- If the coroutine is not yielded from before it is destroyed,
- an error message is logged.
- """
- if _inspect_iscoroutinefunction(func):
- # In Python 3.5 that's all we need to do for coroutines
- # defiend with "async def".
- # Wrapping in CoroWrapper will happen via
- # 'sys.set_coroutine_wrapper' function.
- return func
-
- if inspect.isgeneratorfunction(func):
- coro = func
- else:
- @_wraps(func)
- def coro(*args, **kw):
- res = func(*args, **kw)
- if (isinstance(res, futures._FUTURE_CLASSES)
- or inspect.isgenerator(res)):
- res = yield From(res)
- elif _AwaitableABC is not None:
- # If 'func' returns an Awaitable (new in 3.5) we
- # want to run it.
- try:
- await_meth = res.__await__
- except AttributeError:
- pass
- else:
- if isinstance(res, _AwaitableABC):
- res = yield From(await_meth())
- raise Return(res)
-
- if _PEP479:
- # FIXME: use @_wraps
- coro = pep479_wrapper(func, coro)
- coro = _wraps(func)(coro)
-
- if not _DEBUG:
- if _types_coroutine is None:
- wrapper = coro
- else:
- wrapper = _types_coroutine(coro)
- else:
- @_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 = True # For iscoroutinefunction().
- return wrapper
-
-
-def iscoroutinefunction(func):
- """Return True if func is a decorated coroutine function."""
- return (getattr(func, '_is_coroutine', False) or
- _inspect_iscoroutinefunction(func))
-
-
-_COROUTINE_TYPES = (types.GeneratorType, CoroWrapper)
-if _CoroutineABC is not None:
- _COROUTINE_TYPES += (_CoroutineABC,)
-if events.asyncio is not None:
- # Accept also asyncio CoroWrapper for interoperability
- if hasattr(events.asyncio, 'coroutines'):
- _COROUTINE_TYPES += (events.asyncio.coroutines.CoroWrapper,)
- else:
- # old asyncio/Python versions
- _COROUTINE_TYPES += (events.asyncio.tasks.CoroWrapper,)
-
-def iscoroutine(obj):
- """Return True if obj is a coroutine object."""
- return isinstance(obj, _COROUTINE_TYPES)
-
-
-def _format_coroutine(coro):
- assert iscoroutine(coro)
-
- coro_name = None
- if isinstance(coro, CoroWrapper):
- func = coro.func
- coro_name = coro.__qualname__
- if coro_name is not None:
- coro_name = '{0}()'.format(coro_name)
- else:
- func = coro
-
- if coro_name is None:
- coro_name = events._format_callback(func, ())
-
- try:
- coro_code = coro.gi_code
- except AttributeError:
- coro_code = coro.cr_code
-
- try:
- coro_frame = coro.gi_frame
- except AttributeError:
- coro_frame = coro.cr_frame
-
- filename = coro_code.co_filename
- if (isinstance(coro, CoroWrapper)
- and not inspect.isgeneratorfunction(coro.func)
- and coro.func is not None):
- filename, lineno = events._get_function_source(coro.func)
- if coro_frame is None:
- coro_repr = ('%s done, defined at %s:%s'
- % (coro_name, filename, lineno))
- else:
- coro_repr = ('%s running, defined at %s:%s'
- % (coro_name, filename, lineno))
- elif coro_frame is not None:
- lineno = coro_frame.f_lineno
- coro_repr = ('%s running at %s:%s'
- % (coro_name, filename, lineno))
- else:
- lineno = coro_code.co_firstlineno
- coro_repr = ('%s done, defined at %s:%s'
- % (coro_name, filename, lineno))
-
- return coro_repr
-
-
-class FromWrapper(object):
- __slots__ = ('obj',)
-
- def __init__(self, obj):
- if isinstance(obj, FromWrapper):
- obj = obj.obj
- assert not isinstance(obj, FromWrapper)
- self.obj = obj
-
-def From(obj):
- if not _DEBUG:
- return obj
- else:
- return FromWrapper(obj)
projects / dotfiles / .git / blobdiff