如何从Qu获取事件循环

Question

你好，我对Python相当陌生，我正在尝试将Flask上的一个现有应用程序转换为Quart（https://gitlab.com/pgjones/quart），它应该构建在asyncio之上，这样我就可以使用Goblin OGM与JanusGraph或TinkerPop进行交互。根据我在Goblin上找到的示例，我需要获得一个事件循环来异步运行命令。在

    >>> import asyncio
    >>> from goblin import Goblin

    >>> loop = asyncio.get_event_loop()
    >>> app = loop.run_until_complete(
    ...     Goblin.open(loop))
    >>> app.register(Person, Knows)

但是，我找不到从Quart获取事件循环的方法，即使它是在asyncio之上构建的。在

有人知道我怎么弄到的吗？任何帮助将不胜感激。在

Answer 1

TL；DR要获取事件循环，请调用asyncio.get_event_loop()。在

在基于asyncio的应用程序中，事件循环通常不属于Quart或任何其他协议/应用程序级组件，它由asyncio提供，也可能是像uvloop这样的加速器。事件循环通过调用asyncio.get_event_loop()获得，有时用asyncio.set_event_loop()设置。在

这就是quart的app.run()用来运行应用程序的，这意味着它与asyncio为主线程创建的默认事件循环一起工作。在您的例子中，您只需在注册Goblin后调用quart的run()：

loop = asyncio.get_event_loop()
goblin_app = loop.run_until_complete(Goblin.open(loop))
goblin_app.register(Person, Knows)
quart_app = Quart(...)
# ... @app.route, etc

# now they both run in the same event loop
quart_app.run()

---

以上这些都应该从实际意义上回答这个问题。但是，如果不止一个组件坚持要有自己的run()方法来旋转事件循环，那么这种方法就行不通了——因为app.run()没有返回，你只能在一个线程中调用一个这样的函数。

但是，如果仔细观察，quart也不是这样。虽然Quart示例确实使用app.run()来为应用程序提供服务，但是如果您查看一下app.run()的实现，您将看到它调用了方便函数run_app()，该函数简单地创建了一个服务器并永远启动主循环：

^{pr2}$

如果需要控制事件循环的实际运行方式，则始终可以自己执行：

# obtain the event loop from asyncio
loop = asyncio.get_event_loop()

# hook Goblin to the loop
goblin_app = loop.run_until_complete(Goblin.open(loop))
goblin_app.register(Person, Knows)

# hook Quart to the loop
quart_server = loop.run_until_complete(loop.create_server(
        lambda: quart.serving.Server(quart_app, loop), host, port))

# actually run the loop (and the program)
try:
    loop.run_forever()
except KeyboardInterrupt:  # pragma: no cover
    pass
finally:
    quart_server.close()
    loop.run_until_complete(quart_server.wait_closed())
    loop.run_until_complete(loop.shutdown_asyncgens())
    loop.close()