Python异步类构造器设计模式：避免在__init__中执行异步操作

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

class CosmosCRUD:
    def __init__(self, client: CosmosClient):
        self.client = client
        # 以下代码会导致SyntaxError，因为__init__不是async函数
        # 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={"kind": "Hash", "paths": ["/id"]})

import asyncio
from azure.cosmos.aio import CosmosClient

class CosmosCRUDBad:
    def __init__(self, client: CosmosClient):
        self.client = client
        loop = asyncio.get_event_loop() # 或者创建一个新的loop
        # 这种方式会阻塞当前线程，直到异步操作完成
        self.database = loop.run_until_complete(
            self.client.create_database_if_not_exists("MY_DATABASE_NAME")
        )
        self.container = loop.run_until_complete(
            self.database.create_container_if_not_exists("MY_CONTAINER_NAME", partition_key={"kind": "Hash", "paths": ["/id"]})
        )

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

class CosmosCRUD:
    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, client: CosmosClient, db_name: str, container_name: str, partition_key_path: str = "/id"):
        """
        异步工厂方法，负责所有异步初始化逻辑。
        """
        print(f"正在异步初始化 CosmosCRUD 实例...")
        database = await client.create_database_if_not_exists(db_name)
        container = await database.create_container_if_not_exists(
            container_name, partition_key={"kind": "Hash", "paths": [partition_key_path]}
        )
        # 初始化完成后，调用同步构造函数创建并返回实例
        return cls(client, database, container)

    async def create_item(self, item: dict):
        """
        示例：异步创建文档
        """
        response = await self.container.create_item(item, enable_automatic_id_generation=True)
        print(f"创建项成功: {response['id']}")
        return response

    async def read_item(self, item_id: str, partition_key: str):
        """
        示例：异步读取文档
        """
        response = await self.container.read_item(item_id, partition_key=partition_key)
        print(f"读取项成功: {response['id']}")
        return response

# 如何使用这个类
async def main():
    # 假设 client 已经是一个异步 CosmosClient 实例
    # 通常在应用启动时创建一次
    # client = CosmosClient(url="YOUR_COSMOS_URL", credential="YOUR_COSMOS_KEY")
    # 为了演示，我们使用一个模拟的客户端
    class MockCosmosClient:
        async def create_database_if_not_exists(self, name):
            print(f"模拟：创建或获取数据库 '{name}'")
            return MockDatabaseProxy(name)
    class MockDatabaseProxy:
        def __init__(self, name): self.name = name
        async def create_container_if_not_exists(self, name, partition_key):
            print(f"模拟：创建或获取容器 '{name}' (数据库: {self.name})")
            return MockContainerProxy(name)
    class MockContainerProxy:
        def __init__(self, name): self.name = name
        async def create_item(self, item, enable_automatic_id_generation):
            print(f"模拟：创建项 {item}")
            return {"id": "mock_id_123", **item}
        async def read_item(self, item_id, partition_key):
            print(f"模拟：读取项 {item_id} (分区键: {partition_key})")
            return {"id": item_id, "data": "some_data", "pk": partition_key}

    mock_client = MockCosmosClient()

    # 使用异步工厂方法创建实例
    crud_manager = await CosmosCRUD.create(
        client=mock_client,
        db_name="MyAwesomeDB",
        container_name="MyContainer",
        partition_key_path="/pk"
    )

    # 现在可以使用 crud_manager 实例进行异步操作
    new_item = await crud_manager.create_item({"pk": "category1", "name": "Product A"})
    await crud_manager.read_item(new_item["id"], "category1")

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

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

# 假设 CosmosClient 已经初始化
# cosmos_client = CosmosClient(...)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 应用程序启动时运行
    print("应用启动中...")
    global crud_manager_instance
    # 异步创建 CosmosCRUD 实例
    crud_manager_instance = await CosmosCRUD.create(
        client=cosmos_client,
        db_name="MyAwesomeDB",
        container_name="MyContainer",
        partition_key_path="/pk"
    )
    yield
    # 应用程序关闭时运行
    print("应用关闭中...")
    # 这里可以添加资源清理逻辑，例如关闭 CosmosClient
    # await cosmos_client.close()

app = FastAPI(lifespan=lifespan)

# 依赖函数，提供 CosmosCRUD 实例
async def get_cosmos_crud() -> CosmosCRUD:
    return crud_manager_instance

@app.get("/items/{item_id}")
async def read_item_route(item_id: str, pk: str, crud: CosmosCRUD = Depends(get_cosmos_crud)):
    item = await crud.read_item(item_id, pk)
    return {"message": "Item retrieved", "item": item}

@app.post("/items/")
async def create_item_route(item_data: dict, crud: CosmosCRUD = Depends(get_cosmos_crud)):
    new_item = await crud.create_item(item_data)
    return {"message": "Item created", "item": new_item}