Python异步类构造器的设计模式与最佳实践

import asyncio
from azure.cosmos.aio import CosmosClient # 假设已经安装并导入

class CosmosCRUD:
    def __init__(self, client: CosmosClient):
        self.client = client
        # 以下代码会导致SyntaxError，因为__init__不能await
        # 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="/id")
        print("CosmosCRUD实例已创建（同步部分）")

# 无法直接实例化并期望异步操作完成
# client = CosmosClient(...)
# crud_instance = CosmosCRUD(client) # 无法执行异步初始化

import asyncio
from azure.cosmos.aio import CosmosClient, DatabaseProxy, ContainerProxy # 假设已经安装并导入

# 为了演示，我们模拟CosmosClient的异步行为
class MockCosmosClient:
    async def create_database_if_not_exists(self, db_name: str):
        print(f"模拟：创建或获取数据库 '{db_name}'...")
        await asyncio.sleep(0.1) # 模拟网络延迟
        return MockDatabaseProxy(db_name)

class MockDatabaseProxy:
    def __init__(self, name: str):
        self.name = name

    async def create_container_if_not_exists(self, container_name: str, partition_key: str):
        print(f"模拟：在数据库 '{self.name}' 中创建或获取容器 '{container_name}' (分区键: {partition_key})...")
        await asyncio.sleep(0.1) # 模拟网络延迟
        return MockContainerProxy(container_name)

class MockContainerProxy:
    def __init__(self, name: str):
        self.name = name
        print(f"模拟：容器 '{self.name}' 初始化完成。")

# 实际的CosmosCRUD类
class CosmosCRUD:
    # 声明实例变量的类型，帮助IDE进行类型检查
    client: CosmosClient
    database: DatabaseProxy
    container: ContainerProxy

    def __init__(self, client: CosmosClient, database: DatabaseProxy, container: ContainerProxy):
        """
        同步构造器，只负责赋值，不执行异步操作。
        """
        self.client = client
        self.database = database
        self.container = container
        print("CosmosCRUD实例的同步部分初始化完成。")

    @classmethod
    async def create(cls, cosmos_client: CosmosClient, db_name: str, container_name: str, partition_key: str):
        """
        异步工厂方法，负责所有异步初始化逻辑。
        """
        print(f"开始异步初始化 CosmosCRUD 实例...")
        database = await cosmos_client.create_database_if_not_exists(db_name)
        container = await database.create_container_if_not_exists(container_name, partition_key=partition_key)

        # 调用同步构造器创建实例
        instance = cls(cosmos_client, database, container)
        print("CosmosCRUD实例的异步部分初始化完成。")
        return instance

    async def get_item(self, item_id: str):
        """
        示例：使用已初始化的容器执行CRUD操作
        """
        print(f"从容器 '{self.container.name}' 获取项目 '{item_id}'...")
        await asyncio.sleep(0.05) # 模拟异步操作
        return {"id": item_id, "data": "some_data"}

# 如何使用这个类
async def main():
    # 假设cosmos_client是一个已配置好的异步CosmosClient实例
    # client = CosmosClient.from_connection_string("YOUR_CONNECTION_STRING")

    # 在本例中使用模拟客户端
    mock_client = MockCosmosClient()

    # 使用异步工厂方法创建实例
    crud_instance = await CosmosCRUD.create(
        cosmos_client=mock_client,
        db_name="MY_DATABASE_NAME",
        container_name="MY_CONTAINER_NAME",
        partition_key="/id"
    )

    # 现在可以使用crud_instance进行异步操作
    item = await crud_instance.get_item("123")
    print(f"获取到的项目: {item}")

    # 实际应用中，记得关闭客户端
    # await mock_client.close() 

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