如何在类构造函数中使用异步代码？FastAPI+CosmosDB场景

针对你在异步FastAPI项目中设计Cosmos DB CRUD类构造函数的痛点，这里给出两种可行的解决方案，既满足异步初始化需求，又能避免IDE报错：

方案一：异步工厂方法+显式类型声明 #

通过私有构造函数强制用户使用异步工厂方法创建实例，同时在类级别声明实例变量类型，让IDE识别属性的存在：

from typing import Type, TypeVar
from azure.cosmos.aio import CosmosClient, DatabaseProxy, ContainerProxy

T = TypeVar("T", bound="CosmosCRUD")

class CosmosCRUD:
    # 类级别声明变量类型，告知IDE这些属性会被初始化
    client: CosmosClient
    database: DatabaseProxy
    container: ContainerProxy

    def __init__(self) -> None:
        # 私有构造函数，禁止直接实例化
        self.client = None  # type: ignore
        self.database = None  # type: ignore
        self.container = None  # type: ignore

    @classmethod
    async def create(cls: Type[T], client: CosmosClient) -> T:
        instance = cls()
        instance.client = client
        # 异步初始化数据库和容器
        instance.database = await client.create_database_if_not_exists("MY_DATABASE_NAME")
        instance.container = await instance.database.create_container_if_not_exists(
            "MY_CONTAINER_NAME",
            partition_key="/partitionKey"  # 替换为你的实际分区键
        )
        return instance

# 使用示例
async def init_crud():
    cosmos_client = CosmosClient("your-connection-string")
    crud_instance = await CosmosCRUD.create(cosmos_client)
    # 后续调用CRUD方法

方案优势 #

• 严格遵循异步编程规范，不会阻塞事件循环
• 类级别类型注解彻底解决IDE变量未定义的报错
• 私有构造函数避免用户误用同步方式创建实例

方案二：FastAPI依赖注入适配 #

结合FastAPI的异步依赖注入机制，将CRUD实例的初始化逻辑封装为依赖项，完美适配Web请求场景：

from fastapi import FastAPI, Depends
from azure.cosmos.aio import CosmosClient, DatabaseProxy, ContainerProxy

app = FastAPI()

# 定义CosmosClient的全局依赖
async def get_cosmos_client() -> CosmosClient:
    client = CosmosClient("your-connection-string")
    yield client
    await client.close()

# 定义CosmosCRUD的异步依赖
async def get_cosmos_crud(client: CosmosClient = Depends(get_cosmos_client)) -> "CosmosCRUD":
    crud = CosmosCRUD()
    crud.client = client
    crud.database = await client.create_database_if_not_exists("MY_DATABASE_NAME")
    crud.container = await crud.database.create_container_if_not_exists(
        "MY_CONTAINER_NAME",
        partition_key="/partitionKey"
    )
    return crud

class CosmosCRUD:
    client: CosmosClient
    database: DatabaseProxy
    container: ContainerProxy

    def __init__(self) -> None:
        self.client = None  # type: ignore
        self.database = None  # type: ignore
        self.container = None  # type: ignore

# 路由中使用依赖
@app.get("/items/{item_id}")
async def read_item(item_id: str, crud: CosmosCRUD = Depends(get_cosmos_crud)):
    item = await crud.container.read_item(item_id, partition_key=item_id)
    return item

方案优势 #

• 符合FastAPI生态设计，实例可按需复用（通过use_cache=True控制）
• 无需手动管理实例生命周期，FastAPI自动处理异步初始化
• 同样通过类级别类型注解消除IDE报错

避坑提醒 #

绝对不要在__init__中手动创建事件循环执行异步代码，这会阻塞FastAPI的全局事件循环，导致所有请求停滞，严重影响服务性能。

内容的提问来源于stack exchange，提问作者John