Closing resources with dependency injection

Question

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?

Answer 1

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