pythonresourcesresource-management

Closing resources with dependency injection


I have the following pseudocode:

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 is a large function, and I would like to extract the creation of resources and ResourceUser:

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

By doing so, I lose the option to close the resources correctly.

An optional way to solve it is by creating a close() function in ResourceUser:

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

Unfortunately, this assumes that the ResourceUser "owns" the given resources, and has many potential drawbacks such as preventing the sharing of resources (like a connection pool) between multiple instances.

Counting on the __del__ (akin to RAII) is not an option, especially considering the asynchronous nature of the closing method.

Initializing the resources in a different function and then creating the User in main() will result in plenty of different resources in the return statement, which is rather ugly. Monkey-packing the ResourceUser.close() is also pretty ugly.

Is there any standardized way that does not over-complicate yet achieve the desired result?


Solution

  • A pseudo-code how you can handle this with 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