使用依赖关系注入关闭资源

Closing resources with dependency injection

提问人:Bharel 提问时间:10/28/2023 更新时间:10/30/2023 访问量:30

问:

我有以下伪代码:

class Resource:
    """E.g. Session, connection, file, etc."""
    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(...)  # Plenty of those
    await r.open(...)
    # More resource initialization
    try:
        ResourceUser(r, r2, r3)
    finally:
        await r.close(...) # In practice done with AsyncExitStack
        await r.close(...)

Main 是一个大函数,我想提取资源和 ResourceUser 的创建:

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

这样一来,我就失去了正确关闭资源的选项。

解决此问题的可选方法是在 ResourceUser 中创建一个 close() 函数:

async def close(self):
    await self.resource.close()

不幸的是,这假定 ResourceUser“拥有”给定的资源,并且有许多潜在的缺点,例如阻止在多个实例之间共享资源(如连接池)。

指望(类似于 RAII)不是一种选择,特别是考虑到关闭方法的异步性质。__del__

在不同的函数中初始化资源,然后在 main() 中创建 User,将导致 return 语句中出现大量不同的资源,这是相当丑陋的。猴子包装也很丑陋。ResourceUser.close()

有没有一种标准化的方法既不会过于复杂,又能达到预期的效果?

Python 资源管理

评论

1赞 Andrej Kesely 10/29/2023
您可以使用异步上下文管理器吗?
0赞 Bharel 10/30/2023
@AndrejKesely以什么方式?你的意思是创建初始化,生成创建的资源,然后关闭所有内容?这有点聪明,我喜欢这个解决方案
0赞 Andrej Kesely 10/30/2023
我在这里发布了一个伪代码,如何使用异步上下文管理器。如果可以,您可以接受它并关闭此问题。

答:

1赞 Andrej Kesely 10/30/2023 #1

一个伪代码,你可以用它来处理这个问题contextlib.asynccontextmanager

from contextlib import asynccontextmanager


class Resource:
    """E.g. Session, connection, file, etc."""

    async def open(self):
        pass

    async def close(self):
        pass


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


@asynccontextmanager
async def get_resource_user():
    r1 = Resource(...)  # Plenty of those
    r2 = Resource(...)  # Plenty of those
    try:
        await r1.open(...)  # better use asynccontextmanager for this too, to not call close() explicitly
        await r2.open(...)
        yield ResourceUser(r1, r2)

    finally:
        await r1.close(...)
        await r2.close(...)


async def main():

    async with get_resource_user() as user:
        # do stuff with `user`

    # resources are closed automatically at this point