Scopes
Scopes are one of the fundamental concepts in dependency injection. Some dependency injection frameworks provide fixed scopes, for example:
- Singleton: only one instance is created
- Request: in web frameworks, this could be the lifetime of a request
- Prototype: re-initialized every time it is needed
di
generalizes this concept by putting control of scopes into the hands of the users / implementers: a scope in di
is identified by any hashable value (a string, enum, int, etc.) and entering / exiting scopes is handled via context managers:
async with container.enter_scope("app"):
async with container.enter_scope("request"):
async with container.enter_scope("foo, bar, baz!"):
Scopes provide a framework for several other important features:
- Dependency lifespans
- Dependency value sharing
Every dependency is linked to a scope. When a scope exits, all dependencies linked to it are destroyed (if they have teardown, the teardown is run) and their value is removed from the cache. This means that dependencies scoped to an outer scope cannot depend on dependencies scoped to an inner scope:
from di import Container
from di.dependent import Dependent, Marker
from di.typing import Annotated
class Request:
...
RequestDep = Annotated[Request, Marker(scope="request")]
class DBConnection:
def __init__(self, request: RequestDep) -> None:
...
DBConnDep = Annotated[DBConnection, Marker(scope="app")]
def controller(conn: DBConnDep) -> None:
...
def framework() -> None:
container = Container()
container.solve(Dependent(controller, scope="request"), scopes=["app", "request"])
This example will fail with di.exceptions.ScopeViolationError
because an DBConnection
is scoped to "app"
so it cannot depend on Request
which is scoped to "request"
.
The order of the scopes is determined by the scopes
parameter to Container.solve
.
If you've used Pytest fixtures before, you're already familiar with these rules.
In Pytest, a "session"
scoped fixture cannot depend on a "function"
scoped fixture.
Overriding scopes
You may encounter situations where you don't want to make your users explicitly set the scope for each dependency.
For example, Spring defaults dependencies to the "singleton" scope.
Our approach is to give you a callback that gets information on the current context (the scopes passed to Container.solve
, the current DependentBase
and the scopes of all of it's sub-dependencies) where you can inject your own logic for determining the right scope.
Some examples of this include:
- A fixed default scope. You ignore all of the inputs and return a fixed value. This allows you to emulate Spring's behavior by returning a "singleton" scope or FastAPI's behavior by returning a "connection"/"request" scope.
- Try to assign the outermost valid scope. If the dependency depends on a
"request"
sub-dependency, you can't assign a"singleton"
scope, so you assign the"request"
scope. If there are no sub-dependencies or they all have the"singleton"
scope, then you can assign the"singleton"
scope.
Here is an example of the simpler fixed-default behavior:
import os
from typing import Any, Sequence
from di import Container
from di.api.dependencies import DependentBase
from di.api.scopes import Scope
from di.dependent import Dependent, Marker
from di.executors import AsyncExecutor
from di.typing import Annotated
# Framework code
class Request:
def __init__(self, domain: str) -> None:
self.domain = domain
async def web_framework() -> None:
container = Container()
container.bind(
lambda param, dependent: Dependent(Request, scope="request", wire=False)
if dependent.call is Request
else None
)
def scope_resolver(
dep: DependentBase[Any],
subdep_scopes: Sequence[Scope],
scopes: Sequence[Scope],
) -> Scope:
if dep.scope is None:
return "request"
return dep.scope
solved = container.solve(
Dependent(controller, scope="request"),
scopes=["singleton", "request"],
scope_resolver=scope_resolver,
)
async with container.enter_scope("singleton") as singleton_state:
os.environ["domain"] = "bar.example.com"
async with container.enter_scope(
"request", state=singleton_state
) as request_state:
status = await solved.execute_async(
values={Request: Request("bar.example.com")},
executor=AsyncExecutor(),
state=request_state,
)
assert status == 200, status
os.environ["domain"] = "foo.example.com"
async with container.enter_scope(
"request", state=singleton_state
) as request_state:
status = await solved.execute_async(
values={Request: Request("foo.example.com")},
executor=AsyncExecutor(),
state=request_state,
)
assert status == 200, status
# get_domain_from_env gets the "request" scope
def get_domain_from_env() -> str:
return os.environ["domain"]
# authorize gets the "request" scope
def authorize(
request: Request,
domain: Annotated[str, Marker(get_domain_from_env)],
) -> bool:
return request.domain == domain
async def controller(authorized: Annotated[bool, Marker(authorize)]) -> int:
if authorized:
return 200
return 403
In this example we didn't provide a scope for get_domain_from_env
, but di
can see that it does not depend on anything with the "request"
scope and so it gets assigned the "singleton"
scope.
On the other hand authorize
does depend on a Request
object, so it gets the "request"
scope.