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?
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