python类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()