How to handle resources?

[mario-gazzara]

Hi all,

I come from the dependency injector library, and recently I've been exploring another library to see if it could potentially replace dependency injector altogether.

I find it quite interesting, but I still have some doubts. For instance, how can I manage resource releases with this new library?

Let's consider a scenario where I have a client that needs to be released in a certain time (out of scope, raising an exception or at the end of a job to clean all resources):

def get_sqs_client_gen() -> Generator[SQSClient, None, None]:
    client: Optional[SQSClient] = None

    try:
        client = get_sqs_client()
        yield client
    finally:
        if client:
            client.close()

Is there a provider within this library to handle such resources, or should I manually handle acquisition and release by passing it as a context manager?

Thank you for your support!

[bschnitz]

I unfortunatly didn't have time to look into this myself yet. But there seems to be a similar, already resolved issue covering release of resources: #119

[mario-gazzara]

Yeah, thank you for your reply, I ended up using fastapi-injector even out of the "Fast API context". Thanks to the custom scope that this library offers. I guess it's something similar to flask injector

[jstasiak]

Hey @mario-gazzara,

Yeah, there's no built-in mechanism for what you need here. I'd suggest going the explicit context manager route (or similar) for the time being.

I'm not opposed to introducing a mechanism for this, it's just that I haven't had a need for this in the past so my understanding of the use cases is likely poor and I don't quite have the motivation to do it myself.

@bschnitz linked to something that I think we could put in the library's documentation.

[bschnitz]

@jstasiak: I tried to create a minimal example for the documentation. I'm not sure if I did it correctly. It works, but there may be misunderstandings. Please have a look: #252

[pierec]

The pattern I tend to use for cleaning up resources boils down to providing an [Async]ExitStack singleton:

from contextlib import ExitStack
from dataclasses import dataclass

import injector


@dataclass
class DepA:
    def __post_init__(self):
        print("Initialized:", id(self))

    def cleanup(self):
        print("Cleaned up:", id(self))


@injector.inject
@dataclass
class App:
    a0: DepA
    a1: DepA
    a2: DepA

    def run(self) -> None: ...


class LifecycleModule(injector.Module):
    def configure(self, binder: injector.Binder) -> None:
        binder.bind(ExitStack, ExitStack, scope=injector.SingletonScope)


class SomeOtherModule(injector.Module):
    @injector.provider
    def dep_a(
        self,
        builder: injector.ClassAssistedBuilder[DepA],
        exit_stack: ExitStack,
    ) -> DepA:
        a = builder.build()
        exit_stack.callback(a.cleanup)
        return a


def run_app():
    di = injector.Injector([LifecycleModule, SomeOtherModule])

    with di.get(ExitStack):
        app = di.get(App)
        app.run()


if __name__ == "__main__":
    run_app()

[bschnitz]

Nice example. However I'm using hundreds of classes and heavily depend on autowiring and I can't see how to get that to run with your example. Seems like I would have to implement a provider for every class which needs to have a cleanup. So I'll stick with my method ;).