Add named scoping support without requiring additional functions.

[alexanderlazarev0]

Currently, one can achieve custom scoping in the following way (example with fastAPI):

import contextlib
import random
import typing

from fastapi import Depends, FastAPI

from that_depends import BaseContainer, container_context, providers


Scope = typing.MutableMapping[str, typing.Any]
Message = typing.MutableMapping[str, typing.Any]
Receive = typing.Callable[[], typing.Awaitable[Message]]
Send = typing.Callable[[Message], typing.Awaitable[None]]
ASGIApp = typing.Callable[[Scope, Receive, Send], typing.Awaitable[None]]


async def async_resource() -> typing.AsyncIterator[int]:
    """Async resource."""
    yield random.randint(1, 100)  # noqa: S311


class MyContainer(BaseContainer):
    """Container."""

    app_provider = providers.Resource(async_resource)
    request_provider = providers.ContextResource(async_resource)
    function_provider = providers.ContextResource(async_resource)


@contextlib.asynccontextmanager
async def request_scope() -> typing.AsyncIterator[None]:
    """Enter request scope."""
    async with container_context(MyContainer.request_provider):
        yield


@contextlib.asynccontextmanager
async def function_scope() -> typing.AsyncIterator[None]:
    """Enter function scope."""
    async with container_context(MyContainer.function_provider):
        yield


app = FastAPI()


class MyMiddleware:
    """Middleware."""

    def __init__(
        self,
        app: ASGIApp,
    ) -> None:
        """Init."""
        self.app: typing.Final = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """Call."""
        async with request_scope():
            await self.app(scope, receive, send)


app.add_middleware(MyMiddleware)


async def function() -> int:
    """Get random number."""
    return random.randint(1, 100)  # noqa: S311


@app.get("/")
async def root(
    app_variable: typing.Annotated[int, Depends(MyContainer.app_provider)],
    first_request_variable: typing.Annotated[int, Depends(MyContainer.request_provider)],
) -> dict[str, int]:
    """Root."""
    # always enter function scope before calling a function
    # this can also be achieved by wrapping the @inject decorator
    async with function_scope():
        first_function_variable = await function()
    async with function_scope():
        second_function_variable = await function()
    second_request_variable = await MyContainer.request_provider()

    return {
        "app": app_variable,  # same across requests
        "first_request": first_request_variable,
        "second_request": second_request_variable,  # same as value above
        "first_function": first_function_variable,
        "second_function": second_function_variable,
    }


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app)

As an additional note, using a lifespan to initialize a app scoped provider using a ContextResource is not possible because context is stored in a contextVar and each request enters a new async context. Thus, the only option (at least in fastAPI is to use other providers for app-scoped providers.

Goal of this pull request:

• Address some minor bug fixes with warnings
• Allow for named scoping for ContextResources

Proposed scoping API

• Provider definition:

provider = providers.ContextResource(func, scope=Scopes.REQUEST)

• Entering named scope

async with container_context(scope=Scopes.REQUEST): ...

• Extend DIContextMiddleware to accept scope

• Allow @inject to accept named scope
  Likely in combination with setting the default scope of the @inject decorator to Scopes.Function

@inject(scope=Scopes.Function)
async def func(v = Provide[MyContainer.function_scoped_provider]): ...

Notes

• DO NOT force scope order, entering a name scope simply will make it such that context is "reset" for all ContextResources that reference that named scope.

Related Issues

This will also resolve #139 since function-scoped ContextResources will fullfill the requierements in that issue.

[alexanderlazarev0]

@lesnik512 While extending the ContextResource, I wanted to add a new keyword only parameter scope to the init method:

def __init__(
        self,
        creator: typing.Callable[P, typing.Iterator[T_co] | typing.AsyncIterator[T_co]],
        scope: ContextScope | None = None, # temporary solution
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
...

However, I cannot insert it in between *args and **kwargs as a keyword only argument because it would clash with potential creators that use scope as an argument (also mypy doesn't like it).

So the only viable solution I can see is modifying the Provider API to something like a Builder:

providers.ContextResource(my_func, scope=Scope.APP).with_args(*my_args).with_kwargs(**my_kwargs)

Or by making args and kwargs into proper arguments.

providers.ContextResource(my_func, scope=Scope.APP, args=my_args, kwargs=my_kwargs)

This is of course a breaking API change, but it is necessary if we want to continue to pass more options to Providers, also 2.0.0 has not been released yet so now would be the time to do it.

[alexanderlazarev0]

@lesnik512 These changes make it so that the warning to remove injection only truly gets triggered when literally nothing was injected or overridden. However, this comes at a non-insignificant performance cost because I basically always need to do isinstance checks.

So the other suggestion would be:
Completely remove the warning from the @inject decorator and if someone missuses it that is on them...

[alexanderlazarev0]

@lesnik512 While extending the ContextResource, I wanted to add a new keyword only parameter scope to the init method:

def __init__(
        self,
        creator: typing.Callable[P, typing.Iterator[T_co] | typing.AsyncIterator[T_co]],
        scope: ContextScope | None = None, # temporary solution
        *args: P.args,
        **kwargs: P.kwargs,
    ) -> None:
...

However, I cannot insert it in between *args and **kwargs as a keyword only argument because it would clash with potential creators that use scope as an argument (also mypy doesn't like it).

So the only viable solution I can see is modifying the Provider API to something like a Builder:

providers.ContextResource(my_func, scope=Scope.APP).with_args(*my_args).with_kwargs(**my_kwargs)

Or by making args and kwargs into proper arguments.

providers.ContextResource(my_func, scope=Scope.APP, args=my_args, kwargs=my_kwargs)

This is of course a breaking API change, but it is necessary if we want to continue to pass more options to Providers, also 2.0.0 has not been released yet so now would be the time to do it.

Okay seems like both of the above are not an option because of limitations of pythons type system (mypy is very unhappy) so the only solutions that seems to work reasonable without using typing.Any or similar is:

providers.ContextResource(my_func, scope=Scope.APP).with_spec(*myargs, **my_kwargs)

This also allows to me to remove a LOT of # type: ignore comments from the code so seems to be for the better.

[lesnik512]

@alexanderlazarev0

providers.ContextResource(my_func, scope=Scope.APP).with_spec(*myargs, **my_kwargs) seems ok to me

[alexanderlazarev0]

@alexanderlazarev0

providers.ContextResource(my_func, scope=Scope.APP).with_spec(*myargs, **my_kwargs) seems ok to me

Yes, thought about this one more time and realized that we will have to split the configuration of the provider into two places anyway (whether its with_spec or whatever)...

Realized that keeping the creator and its arguments together is probably for the best, thus now rewrote to:

providers.ContextResource(my_func, *args, **kwargs).with_config(scope=Scope.APP)

This way, all the additional provider parameters can go into the with_config() method.

[alexanderlazarev0]

Still to-do:

• Make with_config into a real builder.
• Implement changes to the DIContextMiddleware
• Set default of the @inject decorator to ContextScope.INJECT
• Write more complex tests to cover all possible use cases.
• Update documentation
• Update docstrings

[alexanderlazarev0]

Alright, so the way I wrote it currently, named scopes do not prevent entering a resource's scope. This, in combination with the fact that a resource only needs to be initialized to be resolved, leads to the following key example:

p = providers.ContextResource(my_resource).with_config(scope=ContextScopes.APP)

await p.async_resolve()  # will raise an exception

async with container_context(p, scope=None):
    assert get_current_scope() is None
    await p.async_resolve()  # will resolve

I think this behavior is unintuitive, therefore I will add additional guards on entering a resource context that check if the current scope is the named scope.

• Add guards when entering a ContextResource context to check for named scope.
• Prevent container_context from attempting to initialize context for resources not in named scope.
• Allow force entering context.
• Make container_context handle scope and context-item collection during __enter__ instead of __init__
• Adjust migration guide
• Adjust scope documentation

[alexanderlazarev0]

@lesnik512 Ready for review

[vrslev]

I think typing.Literal fits better in public API, since it just easier to type (no need to import stuff). LSP still helps

async with container_context(p, scope="app"): ....

[alexanderlazarev0]

I think typing.Literal fits better in public API, since it just easier to type (no need to import stuff). LSP still helps

async with container_context(p, scope="app"): ....

@vrslev And how would you allow custom scopes?

[vrslev]

Literal[…] | str. But that’s another question—I’ve just commented on ContextScopes

[alexanderlazarev0]

Literal[…] | str. But that’s another question—I’ve just commented on ContextScopes

Well, the | str is exactly my point (btw Literal[…] | str is equivalent to just str)! I also thought about this approach, but this makes scopes very error-prone since you are requiring someone to correctly write a string (multiple times, 1 mistake somewhere will lead to "unexpected" behavior), now you could argue they could follow this approach for scopes:

CUSTOM = "CUSTOM"

with container_context(scope=CUSTOM): # problem solved 

But then you still need to import the CUSTOM variable in all other modules?
Also, this doesn't prevent anyone in the codebase or otherwise from doing this:

with container_context(scope="CUSTOM") # this is not considered an error unless someone brings it up in code-review

So the current approach both acts as an auto-complete and helps to prevent mistakes, since people literally can't use whatever they want and must import a correctly typed value.

I guess my point is the way I wrote it encourages "safe" handling of named contexts.

And yes now you need to import stuff, but I think that is a worthy tradeoff.

The only thing I can suggest is the following (perhaps slightly unrelated):

Currently:

class ContextScope:
    """A named context scope."""

    def __init__(self, name: str) -> None:
        """Initialize a new context scope."""
        self._name = name

    @property
    def name(self) -> str:
        """Get the name of the context scope."""
        return self._name

    @override
    def __eq__(self, other: object) -> bool:
        if isinstance(other, ContextScope):
            return self.name == other.name
        return False

    @override
    def __repr__(self) -> str:
        return f"{self.name!r}"


class ContextScopes:
    """Enumeration of context scopes."""

    ANY = ContextScope("ANY")  # special scope that can be used in any context
    APP = ContextScope("APP")  # application scope
    REQUEST = ContextScope("REQUEST")  # request scope
    INJECT = ContextScope("INJECT")  # inject scope

This implementation has the following downside:

• They have very similar names, which can be confusing.

Proposition: Rename

[vrslev]

Nice write-up. Thanks for clarification, I agree now that current implementation is better than my Literal proposal.