Support Resource Factories

[alexanderlazarev0]

Use case:

I want to inject a Resource into my function, however I want the resource to be created each time I call my function.

Current providers:

• Factories: Do not support context managers/iterators.
• Resources: Provides the resolved object as a singleton.
• ContextResources: "Support" this feature.

providers.ContextResource only work within the container_context, however it makes the code messy when each time before you call a function you need to wrap the call with async with container context(): if you aren't already in a container context.

I would like it so that the following is possible:

async def get_resource() -> AsyncIterator[str]:
    print("Creating...")
    yield "Value"
    print("Destroying...")

class MyContainer(BaseContainer):

    resource = providers.ResourceFactory(get_resource)

@inject
async def injected(s: str = Provide[MyContainer.resource]):
    print(s)

async def main():

    await injected() # Creating... Value Destroying...

    await injected() # Creating... Value Destroying...

So that the resource is created when my injected function is called and then destroyed when the function returns.

Perhaps there is also a workaround for this that I am missing?

[lesnik512]

@alexanderlazarev0 Hi, container_context can be used as decorator, like this

class MyContainer(BaseContainer):
    resource = providers.ContextResource(get_resource)

@container_context()
@inject
async def injected(s: str = Provide[MyContainer.resource]):
    print(s)

It will result in the same behaviour:

So that the resource is created when my injected function is called and then destroyed when the function returns.

[alexanderlazarev0]

@lesnik512 Thank you for the fast response. This does simplify things.

However, this only seems to work as intended in an async context.

import typing
from that_depends import BaseContainer, Provide, providers, container_context, inject

def create_sync_resource() -> typing.Iterator[str]:
    # resource initialization
    try:
        print("setup")
        yield "sync resource"
    finally:
        print("teardown")

class DIContainer(BaseContainer):
    context_resource = providers.ContextResource(create_sync_resource)

@container_context()
@inject
def injected(s: str = Provide[DIContainer.context_resource]):
    print(s)

if __name__ == "__main__":
    injected() #RuntimeWarning: coroutine 'injected' was never awaited

So it is not possible to use ContextResource dependencies in a sync context?

Even in an otherwise async context:

# How I would like to call the function
async def my_function():
    injected() #RuntimeWarning: coroutine 'injected' was never awaited

# Workaround solution
@inject
def injected(s: str = Provide[DIContainer.context_resource]):
    print(s)

async def my_function():
    async with container_context():
        injected()

But in the solution, I am once again forced to manage the container_context prior to calling my function...

[lesnik512]

@alexanderlazarev0 Yes, container_context is only suited for async context. I'll think of some universal solution