Python中异步类构造的最佳实践：避免在__init__中使用await-Python教程-PHP中文网

Python中异步类构造的最佳实践：避免在__init__中使用await

在python中，`__init__`方法不能直接包含`await`操作，因为它是同步的。本文将探讨为什么应避免在构造函数中执行异步逻辑，并提供一种推荐的解决方案——使用异步工厂方法模式来初始化需要异步资源（如数据库连接）的类，同时解决ide关于未初始化变量的警告，确保代码的健壮性和可维护性。

异步操作与Python构造函数的限制

Python的__init__方法是一个同步方法，其设计目的是用于初始化对象的状态，而非执行可能需要等待外部I/O的操作。这意味着，任何涉及await关键字的异步操作都无法直接在__init__中执行，否则会引发SyntaxError。

尝试在__init__中直接使用await，例如连接数据库、创建资源等，是常见的初学者困惑。然而，这种做法不仅在语法上不被允许，从软件设计的角度来看也并非最佳实践。一个良好的类构造函数应该能够快速、无副作用地完成对象的实例化，使其成为“可简单构造”的（trivially constructible）。将耗时的异步操作放入构造函数会违反这一原则，可能导致应用程序启动缓慢、资源管理复杂化，甚至在某些并发场景下引发不可预测的问题。

为什么避免在init中执行异步逻辑

语法限制： Python语言规定__init__不能是async def函数，因此无法直接使用await。
设计原则： 构造函数的职责是初始化对象，而不是执行业务逻辑或耗时的I/O操作。一个对象应该在构造完成后立即可用，即使其内部资源尚未完全准备好。
性能影响： 如果在构造函数中强制执行异步操作（例如通过创建新的事件循环），可能会阻塞主事件循环，尤其是在Web框架（如FastAPI）中，这将严重影响请求处理性能。
资源管理： 在构造函数中创建异步资源（如数据库连接）使得资源的正确关闭和生命周期管理变得复杂。

推荐解决方案：异步工厂方法模式

解决在类初始化时执行异步操作的最佳实践是采用异步工厂方法模式。这种模式将异步初始化逻辑从__init__中分离出来，放入一个单独的异步类方法或静态方法中。

核心思想

__init__方法保持同步且轻量，仅用于初始化对象的内部状态（如成员变量的声明）。
创建一个异步的类方法（通常命名为create、build或from_config），它负责执行所有异步初始化步骤。
这个异步工厂方法首先调用__init__来创建对象实例，然后对该实例执行所有必要的异步设置，最后返回完全初始化的实例。

示例代码：使用异步工厂方法初始化CosmosDB客户端

假设我们有一个CosmosCRUD类，需要异步地确保Cosmos DB数据库和容器的存在。

立即学习“Python免费学习笔记（深入）”；

Haiper

一个感知模型驱动的AI视频生成和重绘工具，提供文字转视频、图片动画化、视频重绘等功能

227

查看详情

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

# 假设这些是你的Cosmos DB配置
COSMOS_DB_NAME = "MY_DATABASE_NAME"
COSMOS_CONTAINER_NAME = "MY_CONTAINER_NAME"
# 假设 partition_key 是一个字典或字符串，例如 {"path": "/id"}
COSMOS_PARTITION_KEY = {"path": "/id"} 

class CosmosCRUD:
    """
    用于执行Cosmos DB CRUD操作的异步类。
    使用异步工厂方法进行初始化。
    """

    # 声明实例变量，以便IDE能够识别它们
    # 在__init__中可以赋None，表示它们将在异步工厂方法中被初始化
    client: CosmosClient
    database: DatabaseProxy
    container: ContainerProxy

    def __init__(self, client: CosmosClient):
        """
        同步构造函数，仅用于接收CosmosClient实例。
        不执行任何异步操作。
        """
        self.client = client
        # 可以在这里初始化为None，明确表示它们稍后会被赋值
        self.database = None 
        self.container = None

    @classmethod
    async def create(cls, client: CosmosClient) -> "CosmosCRUD":
        """
        异步工厂方法，负责CosmosCRUD实例的异步初始化。
        确保数据库和容器的存在。
        """
        # 1. 调用同步构造函数创建实例
        instance = cls(client)

        # 2. 执行所有异步初始化逻辑
        # 确保数据库存在
        instance.database = await instance.client.create_database_if_not_exists(COSMOS_DB_NAME)

        # 确保容器存在
        instance.container = await instance.database.create_container_if_not_exists(
            COSMOS_CONTAINER_NAME, 
            partition_key=COSMOS_PARTITION_KEY
        )

        print(f"Cosmos DB '{COSMOS_DB_NAME}' and container '{COSMOS_CONTAINER_NAME}' ensured.")

        # 3. 返回完全初始化的实例
        return instance

    async def create_item(self, item: dict) -> dict:
        """
        示例：创建Cosmos DB文档
        """
        if not self.container:
            raise RuntimeError("CosmosCRUD not properly initialized. Container is missing.")
        response = await self.container.create_item(body=item)
        print(f"Item created: {response}")
        return response

    async def read_item(self, item_id: str, partition_key_value: str) -> Optional[dict]:
        """
        示例：读取Cosmos DB文档
        """
        if not self.container:
            raise RuntimeError("CosmosCRUD not properly initialized. Container is missing.")
        try:
            # 注意：读取操作需要提供partition_key
            response = await self.container.read_item(item=item_id, partition_key=partition_key_value)
            print(f"Item read: {response}")
            return response
        except Exception as e:
            print(f"Error reading item {item_id}: {e}")
            return None

# 假设你已经有了一个CosmosClient实例
# 这是一个模拟的CosmosClient，实际项目中应使用azure.cosmos.aio.CosmosClient
class MockCosmosClient:
    async def create_database_if_not_exists(self, db_name: str):
        print(f"Mock: Ensuring database '{db_name}' exists.")
        return MockDatabaseProxy(db_name)

class MockDatabaseProxy:
    def __init__(self, db_name: str):
        self.db_name = db_name
    async def create_container_if_not_exists(self, container_name: str, partition_key: dict):
        print(f"Mock: Ensuring container '{container_name}' exists with key {partition_key}.")
        return MockContainerProxy(container_name)

class MockContainerProxy:
    def __init__(self, container_name: str):
        self.container_name = container_name
        self._items = {}
    async def create_item(self, body: dict) -> dict:
        import uuid
        item_id = str(uuid.uuid4())
        body["id"] = item_id # Cosmos DB items usually have an 'id'
        self._items[item_id] = body
        print(f"Mock: Created item with id '{item_id}'.")
        return body
    async def read_item(self, item: str, partition_key: str) -> dict:
        # Simplified mock for read_item
        if item in self._items and self._items[item].get(COSMOS_PARTITION_KEY["path"].strip('/')) == partition_key:
            return self._items[item]
        raise Exception("Item not found or partition key mismatch.")

async def main():
    # 实际应用中，这里会是真正的CosmosClient实例
    # from azure.cosmos.aio import CosmosClient
    # cosmos_client = CosmosClient(url=COSMOS_DB_URL, credential=COSMOS_DB_KEY)
    mock_cosmos_client = MockCosmosClient() 

    print("Initializing CosmosCRUD instance...")
    crud_instance = await CosmosCRUD.create(mock_cosmos_client)
    print("CosmosCRUD instance initialized.")

    # 现在可以使用crud_instance进行操作
    new_item = {"name": "Test Item", "description": "This is a test.", "id": "item1", "category": "books"}
    created_item = await crud_instance.create_item(new_item)

    # 假设partition_key_value是"item1" (对应COSMOS_PARTITION_KEY={"path": "/id"})
    read_item = await crud_instance.read_item("item1", "item1")

    # 假设在FastAPI应用中，CosmosClient通常会在应用启动时创建和关闭
    # await mock_cosmos_client.close() # 实际客户端需要关闭

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

登录后复制

解决IDE关于未初始化变量的警告

在上述示例中，我们通过两种方式解决了IDE（如PyCharm）可能发出的关于self.database和self.container未在__init__中初始化的警告：

类型注解（Type Hints）： 在类定义中明确声明实例变量的类型，例如 database: DatabaseProxy。这告诉IDE这些属性将存在且具有特定类型。
__init__中初始化为None： 在__init__中将这些变量显式初始化为None（例如 self.database = None）。这使得变量在对象创建时就存在，只是其值暂时为None，待异步工厂方法完成后才会被赋予实际值。IDE会识别这种模式，并减少误报。

集成到FastAPI等异步框架

在FastAPI这样的异步Web框架中，这种异步工厂模式尤其有用。你可以在应用启动时（例如使用@app.on_event("startup")）创建并初始化这些需要异步资源的类实例，然后通过依赖注入将其提供给路由处理函数。

# FastAPI 示例伪代码
from fastapi import FastAPI, Depends
from azure.cosmos.aio import CosmosClient # 假设是真实的客户端

app = FastAPI()

# 全局变量来存储初始化后的CRUD实例
cosmos_crud_instance: Optional[CosmosCRUD] = None
global_cosmos_client: Optional[CosmosClient] = None

@app.on_event("startup")
async def startup_event():
    global global_cosmos_client, cosmos_crud_instance
    # 实际应用中，这里应根据配置创建CosmosClient
    # 例如：global_cosmos_client = CosmosClient(url=..., credential=...)
    global_cosmos_client = MockCosmosClient() # 使用Mock客户端
    cosmos_crud_instance = await CosmosCRUD.create(global_cosmos_client)
    print("FastAPI startup: CosmosCRUD instance ready.")

@app.on_event("shutdown")
async def shutdown_event():
    global global_cosmos_client
    if global_cosmos_client:
        # 实际客户端需要关闭
        # await global_cosmos_client.close() 
        print("FastAPI shutdown: CosmosClient closed.")

async def get_cosmos_crud() -> CosmosCRUD:
    """依赖注入函数，提供CosmosCRUD实例"""
    if cosmos_crud_instance is None:
        raise RuntimeError("CosmosCRUD instance not initialized.")
    return cosmos_crud_instance

@app.get("/items/{item_id}")
async def get_item(item_id: str, crud: CosmosCRUD = Depends(get_cosmos_crud)):
    # 假设分区键与item_id相同
    item = await crud.read_item(item_id, item_id) 
    if item:
        return {"message": "Item found", "item": item}
    return {"message": "Item not found"}, 404

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

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

登录后复制