如何在构造函数中创建带有异步代码的类？

Question

我有一个与 Cosmos 数据库通信的 fastAPI 项目。 我的所有 fastAPI 路由都是异步的 (

async def ...

)。 我需要一个异步类来执行 Cosmos DB 的 CRUD 操作。 我遇到的问题是找出该类的构造函数。

我希望构造函数：

• 将 CosmosClient 作为参数 (azure.cosmos.aio.CosmosClient)
• 使用客户端确保数据库已创建
• 使用客户端确保容器已创建

类似这样的：

class CosmosCRUD:
    def __init__(self, client: CosmosClient):
        self.client = CosmosClient
        self.database = await self.client.create_database_if_not_exists("MY_DATABASE_NAME")
        self.container = await self.database.create_container_if_not_exists("MY_CONTAINER_NAME", partition_key=...)

不幸的是，你只能在异步函数中使用

await

，并且

__init__

不能是异步的，所以上面的代码不起作用。

据我所知，有一些解决方案：

• 在

  __init__

  内创建一个新的事件循环，并运行获取其中的数据库和容器的异步代码
  • 据我所知，这意味着 fastAPI 将在创建此类实例时停止处理所有请求，这可能会出现在每个请求上。这绝对会破坏整个应用程序的性能
• 忽略

  __init__

  并创建一个新的异步构造函数，如此类的用户需要调用的

  create

  • IDE (PyCharm) 不知道

    create

    是始终会首先调用的构造函数。您尝试在其他方法中使用的在此新构造函数中创建的任何实例变量都将被 IDE 标记为不存在，因为它们不是在

    __init__

  中创建的
  • 我尝试同时拥有一个

    create

    异步类方法（即实际的构造函数）和一个

    __init__

    方法，在该方法中，我刚刚添加了在

    create

    中创建的任何实例变量的声明，其类型（例如：

    self.client: CosmosClient

    ）但IDE 仍然抱怨它们不存在
• 以某种方式使用依赖注入
  • 类似这样的：https://stackoverflow.com/a/65247387/6423456
  • 不幸的是，我的

    __init__

    方法将客户端作为参数，并且需要将其传递给正在注入的函数
  • 在上面引用的示例中，这就像需要将

    a

    参数传递给正在注入的

    async_dep

    。
  • 这可能吗？我猜不是

是否有一种好方法可以在构造函数中包含异步代码，而不会让 IDE 对缺少实例变量感到不满？

Answer 1

您可以从

AsyncMixin

继承您的类并实现

__ainit__

方法，在那里您可以使用异步函数：

import asyncio

class AsyncMixin:
    def __init__(self, *args, **kwargs):
        """
        Standard constructor used for arguments pass
        Do not override. Use __ainit__ instead
        """
        self.__storedargs = args, kwargs
        self.async_initialized = False

    async def __ainit__(self, *args, **kwargs):
        """Async constructor, you should implement this"""

    async def __initobj(self):
        """Crutch used for __await__ after spawning"""
        assert not self.async_initialized
        self.async_initialized = True
        # pass the parameters to __ainit__ that passed to __init__
        await self.__ainit__(*self.__storedargs[0], **self.__storedargs[1])
        return self

    def __await__(self):
        return self.__initobj().__await__()


class CosmosClient():
    pass


class CosmosCRUD(AsyncMixin):
    def __init__(self, client: CosmosClient):
        self.client = client
        super().__init__()    

    async def __ainit__(self):
        await asyncio.sleep(1)
        print("it works!")
        # self.database = await self.client.create_database_if_not_exists("MY_DATABASE_NAME")
        # self.container = await self.database.create_container_if_not_exists("MY_CONTAINER_NAME", partition_key=...)


async def main():
    c = await CosmosCRUD(client=CosmosClient())

if __name__ == "__main__":
    asyncio.run(main())