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

Question

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

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

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

并且，明智地测试——每个模块是否应该定义一个全局变量，如:
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() 获得，当从协程调用时，它保证返回当前正在运行的循环。