基于依赖注入提取资源创建后，如何正确关闭异步资源？

问题背景 #

现有如下异步资源与使用者的伪代码：

class Resource:
    """示例：会话、连接、文件等异步资源"""
    async def open(self):
        pass
    async def close(self):
        pass

class ResourceUser:
    def __init__(self, resource: Resource):
        self.resource = resource

async def main():
    r = Resource(...)  # 存在大量此类资源
    await r.open(...)
    # 更多资源初始化操作
    try:
        ResourceUser(r, r2, r3)
    finally:
        await r.close(...) # 实际中用AsyncExitStack批量处理
        await r.close(...)

由于main函数体量过大，希望将资源创建+ResourceUser实例化的逻辑提取为独立函数，比如：

async def create_user():
    r, r2, r3, ... = Resource(), ...
    await r.open()
    return ResourceUser(r)

但这样会丢失正确关闭资源的途径，且尝试过的几种方案都存在弊端：

• 给ResourceUser加close()方法：强制让使用者“拥有”资源，阻碍连接池等资源的多实例共享；
• 用__del__模拟RAII：异步关闭逻辑无法通过__del__可靠执行；
• 拆分资源初始化到其他函数、在main中创建User：返回值要携带大量资源，代码繁琐；
• 猴子补丁ResourceUser.close()：写法不优雅，维护性差。

需要一种既简洁又符合规范的标准化方案。

标准化解决方案 #

方案一：返回使用者+独立清理函数 #

将创建逻辑封装为函数，返回ResourceUser实例和一个异步清理函数，由调用方负责调用清理。这种方式既分离了创建逻辑，又不转移资源所有权：

async def create_user():
    # 创建并初始化资源
    r = Resource()
    await r.open()
    r2 = Resource()
    await r2.open()
    
    # 定义清理函数，集中处理资源关闭
    async def cleanup():
        await r.close()
        await r2.close()
    
    return ResourceUser(r, r2), cleanup

# 调用示例
async def main():
    user, cleanup = await create_user()
    try:
        # 正常使用ResourceUser实例
        pass
    finally:
        # 手动触发清理，保证资源被正确关闭
        await cleanup()

如果资源数量较多，可以结合AsyncExitStack简化清理逻辑的编写：

from contextlib import AsyncExitStack

async def create_user():
    stack = AsyncExitStack()
    try:
        r = Resource()
        await r.open()
        # 将资源关闭操作注册到栈中
        stack.push_async_callback(r.close)
        
        r2 = Resource()
        await r2.open()
        stack.push_async_callback(r2.close)
        
        # 返回使用者和栈的统一清理方法
        return ResourceUser(r, r2), stack.pop_all().aclose
    except:
        # 创建过程中出错，立即清理已初始化的资源
        await stack.aclose()
        raise

方案二：异步上下文管理器封装 #

利用Python的异步上下文管理器协议（__aenter__/__aexit__），将资源的创建、初始化和清理逻辑全部封装到上下文管理器中，调用方通过async with语法自动处理资源生命周期：

from contextlib import asynccontextmanager

@asynccontextmanager
async def user_context():
    # 创建并初始化资源
    r = Resource()
    await r.open()
    try:
        r2 = Resource()
        await r2.open()
        try:
            # 把ResourceUser实例交给调用方使用
            yield ResourceUser(r, r2)
        finally:
            # 退出时先关闭内层资源
            await r2.close()
    finally:
        # 最后关闭外层资源
        await r.close()

# 调用示例
async def main():
    async with user_context() as user:
        # 在上下文内使用user，退出时自动清理所有资源
        pass

同样，资源数量多时用AsyncExitStack简化代码：

from contextlib import AsyncExitStack, asynccontextmanager

@asynccontextmanager
async def user_context():
    async with AsyncExitStack() as stack:
        r = Resource()
        await r.open()
        stack.push_async_callback(r.close)
        
        r2 = Resource()
        await r2.open()
        stack.push_async_callback(r2.close)
        
        # 所有资源都注册到栈后，返回使用者
        yield ResourceUser(r, r2)

这两种方案都是Python异步编程中的标准化实践：

• 方案一适合需要手动控制清理时机的场景，灵活性更高；
• 方案二利用语法糖自动处理资源生命周期，代码更简洁可读性更强，是异步场景下管理资源的首选方式。

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