Python uvloop 模块，new_event_loop() 实例源码

def run(coro, loop=None):
    async def main_task():
        pycurl_task = aio.ensure_future(curl_loop())
        try:
            r = await coro
        finally:
            pycurl_task.cancel()
            with suppress(aio.CancelledError):
                await pycurl_task
        return r, pycurl_task

    if loop is None:
        loop = uvloop.new_event_loop()
        # loop = aio.get_event_loop()
    aio.set_event_loop(loop)
    loop.set_exception_handler(exception_handler)
    r, _ = loop.run_until_complete(main_task())
    return r

def pytest_configure(config):
    global LOOP_INIT
    loop_name = config.getoption('--loop')
    factory = {
        "aioloop": asyncio.new_event_loop,
    }
    if uvloop is not None:
        factory["uvloop"] = uvloop.new_event_loop

    if loop_name:
        if loop_name not in factory:
            raise ValueError(
                "{name} is not valid option".format(name=loop_name)
            )
        LOOP_INIT = factory[loop_name]
    else:
        LOOP_INIT = factory["aioloop"]

def loop(request, loop_type):
    old_loop = asyncio.get_event_loop()
    asyncio.set_event_loop(None)
    if loop_type == 'uvloop':
        loop = uvloop.new_event_loop()
    elif loop_type == 'tokio':
        import tokio
        policy = tokio.TokioLoopPolicy()
        asyncio.set_event_loop_policy(policy)
        loop = tokio.new_event_loop()
    else:
        loop = asyncio.new_event_loop()

    yield loop

    loop.close()
    asyncio.set_event_loop(old_loop)
    gc.collect()

def get_event_loop(debug=False):
    if not debug:
        import uvloop
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        asyncio.set_event_loop(uvloop.new_event_loop())
    return asyncio.get_event_loop()

def get_event_loop(self, debug=False):
        if not debug:
            asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
            asyncio.set_event_loop(uvloop.new_event_loop())
        AsyncIOMainLoop().install()
        return asyncio.get_event_loop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def retrieve(urls):
    start_time = time.time()
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)
    future = asyncio.ensure_future(check_urls(urls, loop))
    results = loop.run_until_complete(future)
    logger.info('Execution time: %s seconds' % (time.time() - start_time))
    return results

def loop():
    loop = uvloop.new_event_loop()
    # loop = asyncio.new_event_loop() # just for debugging
    asyncio.set_event_loop(loop)
    return loop

def asyncio_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

def uvloop_loop():
    loop = uvloop.new_event_loop()
    yield loop
    loop.close()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def setUp(self):
        self.maxDiff = None
        self.loop = uvloop.new_event_loop()
        asyncio.set_event_loop(None)

        self.app = self.get_app(self.loop)
        self.client = RESTfmTestClient(self.app)
        self.loop.run_until_complete(self.client.start_server())

def loop():
    loop = uvloop.new_event_loop()
    # loop = asyncio.new_event_loop() # just for debugging
    asyncio.set_event_loop(loop)
    return loop

def aiohttp_tarantool_multi(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_tarantool_multi_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_tarantool_multi")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from data_interface.tarantool_driver import TarantoolDriver
    from aiohttp_server.aiohttpmulticonnect import AiohttpTarantoolMulti
    db = TarantoolDriver
    web_server = AiohttpTarantoolMulti(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def aiohttp_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from data_interface.tarantool_driver import TarantoolDriver
    from aiohttp_server.aiohttponeconnect import AiohttpUniversalOne
    db = TarantoolDriver()
    web_server = AiohttpUniversalOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def aiohttp_postgres_pool(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start aiohttp_postgres_pool_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start aiohttp_postgres_pool")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)
    from data_interface.postgres_driver import PostgresDriver
    from aiohttp_server.aiohttponeconnect import AiohttpUniversalOne
    pd = PostgresDriver(loop)
    db = loop.run_until_complete(pd.open())
    web_server = AiohttpUniversalOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def sanic_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start sanic_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start sanic_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from sanic_server.saniconeconnect import SanicTarantoolOne
    from data_interface.tarantool_driver import TarantoolDriver
    db = TarantoolDriver()
    web_server = SanicTarantoolOne(loop=loop, port=port, db=db)
    loop.create_task(web_server.start())
    loop.create_task(web_server.listner_for_statistics())
    return loop

def japronto_tarantool_one(port=None, uvloop_enable=False):
    """TBD."""
    if uvloop_enable:
        logging.info("start japronto_tarantool_one_uvloop")
        if sys.platform == 'win32':
            logging.error("D'oh! UVLoop is not support Windows!")
            sys.exit()
        else:
            import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
    else:
        logging.info("start japronto_tarantool_one")
        if sys.platform == 'win32':
            loop = asyncio.ProactorEventLoop()
        else:
            loop = asyncio.get_event_loop()
        asyncio.set_event_loop(loop)

    from japronto_server import janprontooneconnect
    from data_interface.tarantool_driver import TarantoolDriver
    db = TarantoolDriver()
    janprontooneconnect.init(loop_param=loop, port_param=port, db_driver=db)
    loop.create_task(janprontooneconnect.listner_for_statistics())
    janprontooneconnect.start()
    # Return false for compatibility japronto implementation with other
    # servers
    # Japronto hard coded to create own loop and run_forever it :(
    return False  # It will raise Exeptions

def setUp(self):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)
        self.event_loop = asyncio.get_event_loop()
        super(BaseAIOrchestraTestCase, self).setUp()

def handle(self, *args, **options):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        parse_by_categories(loop=loop)
        #TODO keywords should be fixed, as it has hard coded kws
        parse_by_keywords(loop=loop)

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully done jobs'))

def handle(self, *args, **options):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        tweets_to_db(loop=loop)
        tweets_by_tag_to_db(loop=loop)
        clean_tweet_hashtags(loop=loop)

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully done parsing jobs'))

def handle(self, *args, **options):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        content_if_empty_all(loop=loop)
        update_db_with_cleaned_content(loop=loop)
        empty_sources(loop=loop)

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully done parsing jobs'))

def handle(self, *args, **options):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        do_youtube(loop=loop)
        clean_youtube_text(loop=loop)

        loop.close()

        self.stdout.write(self.style.SUCCESS('Successfully done Youtube jobs'))

def run():
    asyncio.set_event_loop(uvloop.new_event_loop())
    server = app.create_server(host="0.0.0.0", port=7777)
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(server)
    signal(SIGINT, lambda s, f: loop.stop())
    try:
        loop.run_forever()
    except:
        loop.stop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def run():
    asyncio.set_event_loop(uvloop.new_event_loop())
    server = app.create_server(host="0.0.0.0", port=8888)
    loop = asyncio.get_event_loop()
    asyncio.ensure_future(server)
    signal(SIGINT, lambda s, f: loop.stop())
    try:
        loop.run_forever()
    except:
        loop.stop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def custom_event_loop():
    try:
        import uvloop
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        logger = logging.getLogger(__name__)
        logger.debug('Using uvloop')
    except ImportError:
        pass

    return asyncio.get_event_loop()

def new_loop(self):
        return uvloop.new_event_loop()

def new_loop(self):
        return asyncio.new_event_loop()


###############################################################################
# Socket Testing Utilities
###############################################################################

def init(config):
    log.debug("Loglevel set to %s", logging.getLevelName(log.getEffectiveLevel()))
    asyncio.set_event_loop(None)
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)

    app = create_app(loop, config)
    ssl_context = create_ssl_context(app['config'])
    run_app(app, loop, ssl_context=ssl_context)

def test_uvloop(self):
        loop = uvloop.new_event_loop()
        asyncio.set_event_loop(loop)

        async def uvwait(x):
            assert isinstance(asyncio.get_event_loop(), uvloop.Loop)
            result = await wait(x)
            return result

        result = loop.run_until_complete(uvwait(1))
        self.assertEqual(1, result)

def set_up_server():
    global loop, server
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)
    register('echo', echo)
    register('echo_delayed', echo_delayed)
    register('raise_error', raise_error)
    coro = asyncio.start_server(serve, HOST, PORT)
    server = loop.run_until_complete(coro)

def run_sum_server():
    def sum(x, y):
        return x + y

    aiorpc.register('sum', sum)
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)
    coro = asyncio.start_server(aiorpc.serve, 'localhost', 6000, loop=loop)
    loop.run_until_complete(coro)
    loop.run_forever()

def call():
    async def do(cli):
        for i in range(NUM_CALLS):
            await cli.call('sum', 1, 2)
            # print('{} call'.format(i))
    client = aiorpc.RPCClient('localhost', 6000)
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)
    start = time.time()

    loop.run_until_complete(do(client))

    print('call: %d qps' % (NUM_CALLS / (time.time() - start)))

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def get_event_loop(self):
        import asyncio
        return asyncio.new_event_loop()

def get_event_loop(self):
        import uvloop
        return uvloop.new_event_loop()

def _init_loop():
    try:
        import uvloop
    except ImportError:
        return asyncio.get_event_loop()
    else:
        return uvloop.new_event_loop()

def __init__(self, name, path=None,
                 template_inputs=None,
                 logger=None,
                 event_loop=None,
                 enable_rollback=False):
        """
        Represents AIOrchestra deployment context designed to
        manage deployment through its lifecycle

        :param name: deployment context name
        :type name: str
        :param path: path to TOSCA template
        :type path: str
        :param template_inputs: TOSCA template input parameters
        :type template_inputs: dict
        :param logger: python logger instance
        :param event_loop: asyncio or any compatible event loop
        :type event_loop: asyncio.Loop
        :param enable_rollback: weather to enable rollback on failure or not
        """
        self.__name = name
        self._tmplt = tosca_template.ToscaTemplate(
            path=path, a_file=True,
            parsed_params=template_inputs)
        self.__path = path
        self.origin_nodes = self._tmplt.graph.nodetemplates
        self.vertices = self._tmplt.graph.vertices
        self.inputs_definitions = self._tmplt.inputs
        self.__outputs = self._tmplt.outputs
        self.template_inputs = template_inputs if template_inputs else {}
        self.__status = self.PENDING
        if not logger:
            self.logger = log.UnifiedLogger(
                log_to_console=True,
                level="DEBUG").setup_logger(__name__)
        else:
            self.logger = logger
        if not event_loop:
            if uvloop:
                uv_loop = uvloop.new_event_loop()
                asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
                asyncio.set_event_loop(uv_loop)
            self.event_loop = asyncio.get_event_loop()
        else:
            self.event_loop = event_loop
        self.__orchestra_nodes = [node.OrchestraNode(self, origin_node)
                                  for origin_node in self.origin_nodes]
        self.__deployment_plan = None
        self.rollback_enabled = enable_rollback

def main():
    asyncio.set_event_loop(None)

    loop = uvloop.new_event_loop()

    handler = Handler(loop=loop)
    _root = os.path.abspath(os.path.dirname(__file__))
    handler.lookup_files(os.path.join(_root, 'data'))

    app = web.Application(loop=loop)

    setup_routes(app, handler)

    handler = app.make_handler(access_log=None)

    server = loop.create_server(
        handler,
        os.environ.get('HOST', '0.0.0.0'),
        int(os.environ.get('PORT', 5000)),
    )

    url = URL('https://fake-useragent.herokuapp.com/')

    _heartbeat = loop.create_task(heartbeat(url, 10, 60, loop=loop))

    srv = loop.run_until_complete(server)

    signal.signal(signal.SIGTERM, _sigint)

    try:
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            pass

        _heartbeat.cancel()

        try:
            loop.run_until_complete(_heartbeat)
        except asyncio.CancelledError:
            pass

        srv.close()

        loop.run_until_complete(srv.wait_closed())

        loop.run_until_complete(app.shutdown())

        loop.run_until_complete(handler.finish_connections(5.0))

        loop.run_until_complete(app.cleanup())

    finally:
        loop.call_soon(loop.stop)
        loop.run_forever()
        loop.close()