python asyncio/aiohttp 跨项目共享全局变量

Question

我有一个异步项目。它有几个模块。他们中的许多人需要访问一些全局变量，例如： 1. aiohttp ClientSession() 对象，因为根据 aiohttp 文档，我应该避免为每个请求创建一个新的 ClientSession。
2. asyncio 套接字，即我使用asyncio.open_connection() 创建的reader, writer。我想保持持久连接。
3. 事件循环，我使用asyncio.get_event_loop()

共享此类变量的最佳做法是什么？
我想创建一个globals.py 模块，它将定义这些变量。

问题是我不能在 globals 模块中对 ClientSession 对象使用 async with 语法。
对于套接字，我必须以某种方式在 async def 中定义它，所以我不能在模块级别公开它。

而且，明智的测试 - 每个模块都应该定义一个全局变量，例如：
loop = asyncio.get_event_loop()
或者，将事件循环传递给模块是否更好，例如在类__init__)？

Answer 1

没有必要使用全局变量。相反，创建一个存储“全局”数据的类，并在该类上创建您的函数方法。例如：

class Operation:
    def __init__(self):
        self._http = aiohttp.ClientSession()

    async def open(self):
        self._reader, self._writer = \
            await asyncio.open_connection(<host>, <port>)

    # ... more methods ...

    async def close(self):
        await self._http.close()
        self._writer.close()
        await self._writer.wait_closed()

这有几个优点：

• 它不需要任何全局状态；
• 您可以完全控制状态的构建和拆除位置（一次完成）；
• 如果您决定需要并行完成多个全局操作，这样做很容易 - 只需创建多个 Operation 实例。
• 测试时，您可以根据需要简单地创建和拆除Operation 实例。

有了这些，你的 main 协程可以如下所示：

async def main():
    op = Operation()
    await op.open()
    try:
        await op.do_something()
        ...
    finally:
        await op.close()

asyncio.run(main())
#or asyncio.get_event_loop().run_until_complete(main())

请注意，事件循环不存储在对象上，或以任何方式传递给它。这是因为事件循环始终可以通过asyncio.get_event_loop() 获得，当从协程调用时，可以保证返回当前正在运行的循环。