Python异步类构造器的最佳实践：避免__init__中的异步操作

from azure.cosmos.aio import CosmosClient # 假设这是一个异步客户端

class CosmosCRUD:
    def __init__(self, client: CosmosClient):
        self.client = client
        # 以下代码会导致语法错误，因为__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=...)

from azure.cosmos.aio import CosmosClient, DatabaseProxy, ContainerProxy
import asyncio

class CosmosCRUD:
    """
    Cosmos DB CRUD操作类，采用异步工厂模式进行初始化。
    """
    def __init__(self, client: CosmosClient, database_name: str, container_name: str, partition_key_path: str):
        """
        同步构造器，仅用于声明和存储基本属性。
        异步资源在此处声明为None，等待异步工厂方法进行初始化。
        """
        self.client: CosmosClient = client
        self.database_name: str = database_name
        self.container_name: str = container_name
        self.partition_key_path: str = partition_key_path
        self.database: DatabaseProxy = None  # 声明类型，但初始为None
        self.container: ContainerProxy = None # 声明类型，但初始为None
        print("CosmosCRUD: __init__ called (synchronous part)")

    @classmethod
    async def create(cls, client: CosmosClient, database_name: str, container_name: str, partition_key_path: str):
        """
        异步工厂方法，负责创建实例并执行异步初始化逻辑。
        """
        # 1. 调用同步__init__创建实例
        instance = cls(client, database_name, container_name, partition_key_path)
        # 2. 执行异步初始化
        await instance._async_initialize_resources()
        return instance

    async def _async_initialize_resources(self):
        """
        内部异步初始化方法，处理所有异步资源创建和设置。
        """
        print(f"CosmosCRUD: _async_initialize_resources - Ensuring database '{self.database_name}' exists...")
        self.database = await self.client.create_database_if_not_exists(self.database_name)

        print(f"CosmosCRUD: _async_initialize_resources - Ensuring container '{self.container_name}' exists...")
        self.container = await self.database.create_container_if_not_exists(
            self.container_name, partition_key=self.partition_key_path
        )
        print("CosmosCRUD: Asynchronous resources initialized successfully.")

    async def create_item(self, item: dict):
        """示例：创建Cosmos DB文档"""
        if not self.container:
            raise RuntimeError("Cosmos container not initialized.")
        print(f"CosmosCRUD: Creating item with id '{item.get('id')}'...")
        return await self.container.create_item(item)

    async def read_item(self, item_id: str, partition_key_value: str):
        """示例：读取Cosmos DB文档"""
        if not self.container:
            raise RuntimeError("Cosmos container not initialized.")
        print(f"CosmosCRUD: Reading item '{item_id}'...")
        return await self.container.read_item(item_id, partition_key=partition_key_value)

    async def close(self):
        """清理资源，例如关闭CosmosClient连接"""
        if self.client:
            print("CosmosCRUD: Closing CosmosClient...")
            await self.client.close()
            self.client = None # 清理引用
            print("CosmosClient closed.")

# --- 模拟CosmosClient及其相关代理类以便于运行示例 ---
class MockContainerProxy:
    def __init__(self, name):
        self.name = name
    async def create_item(self, item):
        print(f"Mock: Created item {item.get('id')} in container '{self.name}'")
        return item
    async def read_item(self, item_id, partition_key):
        print(f"Mock: Reading item '{item_id}' from container '{self.name}' (pk: {partition_key})")
        return {"id": item_id, "data": "mock_data", "pk": partition_key}

class MockDatabaseProxy:
    def __init__(self, name):
        self.name = name
    async def create_container_if_not_exists(self, name, partition_key):
        print(f"Mock: Creating/getting container '{name}' (partition_key: {partition_key})")
        return MockContainerProxy(name)

class MockCosmosClient:
    async def create_database_if_not_exists(self, name):
        print(f"Mock: Creating/getting database '{name}'")
        return MockDatabaseProxy(name)
    async def close(self):
        print("Mock: CosmosClient closed.")

# --- 示例用法 ---
async def main():
    print("\n--- Starting main execution ---")
    mock_client = MockCosmosClient()

    # 使用异步工厂方法创建CosmosCRUD实例
    print("\nAttempting to create CosmosCRUD instance...")
    cosmos_crud_instance = await CosmosCRUD.create(
        client=mock_client,
        database_name="MyFastAPIDB",
        container_name="UserItems",
        partition_key_path="/id"
    )
    print("CosmosCRUD instance successfully created and initialized.")

    # 现在可以使用实例的异步方法
    new_item = {"id": "user123", "name": "Alice", "city": "New York"}
    await cosmos_crud_instance.create_item(new_item)

    retrieved_item = await cosmos_crud_instance.read_item("user123", "user123")
    print(f"Retrieved item: {retrieved_item}")

    # 清理资源
    await cosmos_crud_instance.close()
    print("--- Main execution finished ---")

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

from fastapi import FastAPI, Depends
# ... 其他导入和 CosmosCRUD 类定义 ...

app = FastAPI()

# 异步依赖项，用于创建并提供CosmosCRUD实例
async def get_cosmos_crud_instance():
    mock_client = MockCosmosClient() # 实际应用中会是真实的CosmosClient
    cosmos_crud = await CosmosCRUD.create(
        client=mock_client,
        database_name="MyFastAPIDB",
        container_name="UserItems",
        partition_key_path="/id"
    )
    try:
        yield cosmos_crud
    finally:
        await cosmos_crud.close() # 确保资源被清理

@app.get("/users/{user_id}")
async def read_user(user_id: str, cosmos: CosmosCRUD = Depends(get_cosmos_crud_instance)):
    user = await cosmos.read_item(user_id, user_id)
    return {"user": user}

# 运行FastAPI应用: uvicorn your_module_name:app --reload