Python inspect 模块，iscoroutine() 实例源码

def next_tick(self, callback, *args, **kwargs):
        """
        Once the current event loop turn runs to completion,
        call the callback or coroutine function::

            loop.next_tick(callback)


        """
        if inspect.iscoroutinefunction(callback):
            raise Exception("Did you mean ot create a coroutine (got: {})".format(callback))

        if not inspect.iscoroutine(callback):
            callback = partial(callback, *args, **kwargs)

        self.ready[callback] = None

def tick(self):

        self.handle_exceptions()

        if not self.ready:
            # Don't let idle block the loop from exiting
            # There should be other handdles blocking exiting if
            # there is nothing ready
            self.unref_ticker()
            return
        coroutine_or_func, future = self.ready.popitem()

        if inspect.iscoroutine(coroutine_or_func):
            process_until_await(self, future, coroutine_or_func)
        else:
            coroutine_or_func()

def crawl(cls, spiders, response, *, middleware=None):
        try:
            if middleware:
                await cls._handle_input(response, middleware)
            res = []
            for spider in spiders:
                for r in spider.parse(response):
                    if inspect.iscoroutine(r):
                        r = await r
                    if r:
                        res.append(r)
            if middleware:
                res = await cls._handle_output(response, res, middleware)
        except Exception as e:
            try:
                if middleware:
                    await cls._handle_error(response, e, middleware)
            except Exception:
                pass
            return ()
        else:
            return res

def start_requests(cls, spiders, start_urls, *, middleware=None):
        try:
            res = []
            for spider in spiders:
                for r in spider.start_requests(start_urls):
                    if inspect.iscoroutine(r):
                        r = await r
                    if r:
                        res.append(r)
            if middleware:
                res = await cls._handle_start_requests(res, middleware)
        except Exception:
            log.warning("Unexpected error occurred when generating start requests", exc_info=True)
            return ()
        else:
            return res

def crawl(cls, spiders, response, *, middleware=None):
        try:
            if middleware:
                await cls._handle_input(response, middleware)
            res = []
            for spider in spiders:
                for r in spider.parse(response):
                    if inspect.iscoroutine(r):
                        r = await r
                    if r:
                        res.append(r)
            if middleware:
                res = await cls._handle_output(response, res, middleware)
        except Exception as e:
            try:
                if middleware:
                    await cls._handle_error(response, e, middleware)
            except Exception:
                pass
            return ()
        else:
            return res

def start_requests(cls, spiders, start_urls, *, middleware=None):
        try:
            res = []
            for spider in spiders:
                for r in spider.start_requests(start_urls):
                    if inspect.iscoroutine(r):
                        r = await r
                    if r:
                        res.append(r)
            if middleware:
                res = await cls._handle_start_requests(res, middleware)
        except Exception:
            log.warning("Unexpected error occurred when generating start requests", exc_info=True)
            return ()
        else:
            return res

def __getattr__(self, item):
        attr = getattr(self.delegate, item)

        if inspect.iscoroutinefunction(attr) or hasattr(attr,
                                                        "_is_coroutine") and attr._is_coroutine or inspect.iscoroutine(
            attr):
            async def wrapper(*args, **kwargs):
                return self._wrap(await attr(*args, **kwargs))

            return wrapper() if inspect.iscoroutine(attr) else wrapper
        elif inspect.isgeneratorfunction(attr) or inspect.isgenerator(attr):
            def wrapper(*args, **kwargs):
                for entry in attr(*args, **kwargs):
                    yield self._wrap(entry)

            return wrapper if inspect.isgeneratorfunction(attr) else wrapper()
        elif inspect.isfunction(attr):
            def wrapper(*args, **kwargs):
                return self._wrap(attr(*args, **kwargs))

            return wrapper
        else:
            return self._wrap(attr)

def __call__(self, **kwargs):
        """Prepare dependencies and call this Task"""
        start = datetime.datetime.now()
        logger.info("Start task %s", self)
        try:
            if self.plugin_class:
                result = self.func(get_plugin_instance(self.plugin_class), **kwargs)

            else:
                result = self.func(**kwargs)

            if inspect.iscoroutinefunction(self.func):
                assert inspect.iscoroutine(
                    result), "The task is defined as a coroutine function but does not return a coroutine"
                loop = asyncio.get_event_loop()
                if loop.is_closed():
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                result = loop.run_until_complete(result)
                loop.close()
        finally:
            stop = datetime.datetime.now()
            logger.info("End task %s after %s", self, stop - start)

        return result

def __await__(self):
        def exec_when_ready():
            event.wait()
            coro.send(None)

            if not self.exited:
                raise RuntimeError('attempted to "await" in a worker thread')

        if self.exited:
            # This is run in the worker thread
            yield
        else:
            # This is run in the event loop thread
            previous_frame = inspect.currentframe().f_back
            coro = next(obj for obj in gc.get_referrers(previous_frame.f_code)
                        if inspect.iscoroutine(obj) and obj.cr_frame is previous_frame)
            event = Event()
            loop = get_event_loop()
            future = loop.run_in_executor(self.executor, exec_when_ready)
            next(future.__await__())  # Make the future think it's being awaited on
            loop.call_soon(event.set)
            yield future

def is_awaitable(obj):
        # There is no single method which can answer in any case, should wait or not - so need to create one
        # for the suspected cases : func, coro, gen-coro, future,
        #                           class with sync __call__, class with async __call__,
        #                           sync method, async method
        if inspect.isawaitable(obj) or inspect.iscoroutinefunction(obj) or inspect.iscoroutine(obj):
            return True
        elif inspect.isgeneratorfunction(obj):
            return True
        elif CallChain.is_user_defined_class(obj):
            if hasattr(obj, '__call__'):
                return CallChain.is_awaitable(obj.__call__)
            return False
        else:
            return False

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def _is_coroutine(func):
    if sys.version_info > (3, 5, 0):
        return inspect.iscoroutine(func)
    return False

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 iscoroutine(obj):
    """Return True if obj is a coroutine object."""
    return isinstance(obj, _COROUTINE_TYPES)

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 iscoroutine(obj):
    """Return True if obj is a coroutine object."""
    return isinstance(obj, _COROUTINE_TYPES)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def AWAIT(coro):
    '''
    Await for a coroutine in an asynchronous thread.  If coro is
    not a proper coroutine, this function acts a no-op, returning coro.
    '''
    if not iscoroutine(coro):
        return coro

    if hasattr(_locals, 'thread'):
        return _locals.thread.AWAIT(coro)
    else:
        raise errors.AsyncOnlyError('Must be used as async')

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def contains_coro(self) -> 'boolean.Boolean':
        return _coconut_tail_call(self.exists, inspect.iscoroutine)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def __call__(self, *args, **kwargs):
        result = self.orig_method(*args, **kwargs)
        if isinstance(result, GeneratorType) or iscoroutine(result):
            raise TypeError("Generator and coroutine test methods should be"
                            " decorated with tornado.testing.gen_test")
        elif result is not None:
            raise ValueError("Return value from test method ignored: %r" %
                             result)

def sync(*func, timeout=None):
    """
    coroutine decorator, convert a coroutine into a syncronous function::

        @sync(timeout=2)
        async def main(sleep_for):
            await uvio.sleep(sleep_for)
            return 'main returned ok!'

        print(main(1))

    """
    if not func:
        return partial(sync, timeout=timeout)

    func = func[0]

    @wraps(func)
    def inner(*args, **kwargs):
        loop = Loop.create(func.__name__)

        coro = func(*args, **kwargs)

        if not iscoroutine(coro):
            raise Exception("{} is not a coroutine (returned from {})".format(coro, func))

        loop.next_tick(coro)

        if timeout:
            def stop_loop():
                loop.stop()
                raise Exception("timeout")


            timer = loop.set_timeout(stop_loop, timeout)
            # Don't wait for the timout to exit the loop
            timer.unref()

        loop.run()
        loop.close()

        if timeout:
            timer.close()

        if coro.cr_await is not None:
            coro.throw(Exception('coroutine {} should not be running at the end of the loop'.format(coro)))

        # This should not happend
        assert not loop._awaiting, loop._awaiting
        assert not loop.ready, loop.ready

    return inner

def invoke_by_args():
  import argparse
  import asyncio
  import coloredlogs
  import inspect
  import pprint
  coloredlogs.install(fmt='[%(levelname).1s %(asctime)s %(module)s:%(lineno)d] %(message)s',
                      datefmt='%y%m%d %H:%M:%S')
  parser = argparse.ArgumentParser()
  subparsers = parser.add_subparsers(dest='')
  for name, method in _methods.items():
    subparser = subparsers.add_parser(name)
    argcount = method.__code__.co_argcount
    num_defaults = len(method.__defaults__) if method.__defaults__ else 0
    argoffset = argcount - num_defaults
    for index, argname in enumerate(method.__code__.co_varnames[:argcount]):
      if index < argoffset:
        subparser.add_argument(argname, type=method.__annotations__[argname])
      elif argname in method.__annotations__:
        subparser.add_argument(argname,
                               type=method.__annotations__[argname],
                               nargs='?',
                               default=method.__defaults__[index - argoffset])
  args = parser.parse_args(options.leftovers)
  name = getattr(args, '')
  delattr(args, '')
  if not name:
    parser.print_help()
  else:
    loop = asyncio.get_event_loop()
    loop.run_until_complete(db.init())
    try:
      result = _methods[name](**vars(args))
      if inspect.iscoroutine(result):
        result = loop.run_until_complete(result)
      if options.pretty:
        print_func = pprint.pprint
      else:
        print_func = lambda x: print(x) if x is not None else None
      if hasattr(result, '__aiter__'):
        async def aloop():
          async for entry in result:
            print_func(entry)

        loop.run_until_complete(aloop())
      else:
        print_func(result)
    except KeyboardInterrupt:
      pass
    finally:
      loop.set_exception_handler(lambda loop, context: None)

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 = '{}()'.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
    lineno = 0
    if (isinstance(coro, CoroWrapper) and
            not inspect.isgeneratorfunction(coro.func) and
            coro.func is not None):
        source = events._get_function_source(coro.func)
        if source is not None:
            filename, lineno = source
        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

def invoke_by_args():
    import argparse
    import asyncio
    import inspect
    import pprint
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest='')
    for name, method in _methods.items():
        subparser = subparsers.add_parser(name)
        argcount = method.__code__.co_argcount
        num_defaults = len(method.__defaults__) if method.__defaults__ else 0
        argoffset = argcount - num_defaults
        for index, argname in enumerate(method.__code__.co_varnames[:argcount]):
            if index < argoffset:
                subparser.add_argument(argname, type=method.__annotations__[argname])
            elif argname in method.__annotations__:
                subparser.add_argument(argname,
                                       type=method.__annotations__[argname],
                                       nargs='?',
                                       default=method.__defaults__[index - argoffset])
    args = parser.parse_args(options.leftovers)
    name = getattr(args, '')
    delattr(args, '')
    if not name:
        parser.print_help()
    else:
        loop = asyncio.get_event_loop()
        try:
            result = _methods[name](**vars(args))
            if inspect.iscoroutine(result):
                result = loop.run_until_complete(result)
                pass
            if options.options.pretty:
                print_func = pprint.pprint
            else:
                print_func = lambda x: print(x) if x is not None else None
            if hasattr(result, '__aiter__'):
                async def aloop():
                    async for entry in result:
                        print_func(entry)

                loop.run_until_complete(aloop())
            else:
                print_func(result)
        except KeyboardInterrupt:
            pass
        loop.set_exception_handler(lambda t_loop, context: None)