Custom dependency injection mechanism

[ReznikovRoman]

Ask your question...
At our company we use Dependency Injector for DI. To inject dependencies into functions it uses wiring mechanism that requires placing some decorators and specifying default arguments.

Starlite validates all controller's parameters, so I can't simply put my "dependency" there.

Here is an example with DI container (requires dependency-injector):

from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide

from starlite import Starlite, get, Dependency


class Cache:
    def get(self): print("Value")


class Container(containers.DeclarativeContainer):
    cache = providers.Factory(Cache)


@get("/1")
def first(dependency: int = Dependency(default=1, skip_validation=True)) -> None:
    # how does that work?
    print(type(dependency))  # >>> <class 'ellipsis'>


@get("/2")
@inject
def second(cache: Cache = Provide[Container.cache]) -> None:  # how to wrap `Provide` object properly?
    cache.get()


@inject  # wiring example
def on_startup(*args, cache: Cache = Provide[Container.cache], **kwargs) -> None:
    cache.get()


container = Container()
container.wire(modules=[__name__])
app = Starlite(
    route_handlers=[first, second],
    on_startup=[on_startup],  # >>> "Value"
)

So the question is: are there any workarounds for this (for skipping validation and passing a callable as an argument)? Currently Starlite's Dependency doesn't suit me, because it accepts only default, and skip_validation works a little bit strange (first controller in the example). Ideally I'd like to have something similar to FastAPI's Depends (here is an example with DI).

[Goldziher]

There is the Dependency keyword which allows skipping validation: https://starlite-api.github.io/starlite/usage/6-dependency-injection/5-the-dependency-function/

[ReznikovRoman]

We use a lot of features from Dependency Injector that Starlite doesn't have, that's why we can't fully "migrate". So we need to somehow integrate our DI framework with Starlite.

I've already found some workarounds, but I was wondering if Starlite has something that allows to "skip" all those validation rules

[Goldziher]

not at present. It would be good to understand what needs to be skipped and where. Also, what we dont support at the moment that you guys need.

[peterschutt]

The Dependency(skip_validation=True) default value for a function parameter causes that parameter to be typed as Any on the pydantic model that is generated to represent the signature of the handler function. Further, marking a parameter as an explicit dependency will cause an exception to be raised if the dependency isn't provided, or there is no default value provided for the dependency.

Looking at your second "workaround" example - the async handler one - can you store the Container instance on the application State and access that in the get_cache() function that you use to provide the cache dependency? Further, what are you losing by using starlite.Provide in the async example instead of dependency_injector.Provide in the sync example?

[ReznikovRoman]

@peterschutt so with dependency_injector.Provide I don't have to create functions like get_cache (that just returns a "provided" value from DI container). At that moment, that's all I've found :)

But I can't be sure if it is the only "drawback" as we haven't fully migrated yet.

[peterschutt]

Cool well keep us updated - @Goldziher is the @inject decorator something you'd like to try to support?

Also, @ReznikovRoman this is a bug:

@get("/1")
def first(dependency: int = Dependency(default=1, skip_validation=True)) -> None:
    # how does that work?
    print(type(dependency))  # >>> <class 'ellipsis'>

Thanks for bringing it up!

[ReznikovRoman]

As a possible workaround, one can use class-based controllers and inject dependencies as class attributes.

Example:

from dependency_injector import containers, providers
from dependency_injector.wiring import Provide

from starlite import Controller, Starlite, get


class Cache:
    def get(self):
        print("Value")


class Container(containers.DeclarativeContainer):
    cache = providers.Factory(Cache)


class Resource(Controller):
    cache = Provide[Container.cache]

    @get("/1")
    def controller(self) -> None:
        self.cache.get()


container = Container()
container.wire(modules=[__name__])
app = Starlite(route_handlers=[Resource])

I also want to exclude this field from OpenAPI schema, remove all type validations and pass there a callable as a default argument (not a value). Looks like just using Dependency is not enough (as I've listed in some examples).

And my initial question was more about how to make a controller's parameter act like a "normal" python function's parameter

@Goldziher But I'm still curious if you can do something like that in Starlite

[ReznikovRoman]

Check this answer if you need asynchronous injections (or this one for explanations)

[ReznikovRoman]

Another solution is to use Starlite DI mechanism instead of wiring from Dependency Injector. You might want to follow this approach if you need asynchronous injections (asynchronous wiring works only for async def functions, so this answer won't work).

Example:

from functools import partial
from typing import AsyncIterator, TypeAlias

from dependency_injector import containers, providers
from dependency_injector.containers import DynamicContainer

from starlite import Provide, Starlite, get


# cache.py
class Cache:
    @classmethod
    async def from_url(cls) -> "Cache":
        return cls()

    async def get(self) -> None:
        print("Value")


async def init_cache() -> AsyncIterator[Cache]:
    cache = await Cache.from_url()
    yield cache


# containers.py
class Container(containers.DeclarativeContainer):
    cache = providers.Resource(init_cache)


# controllers.py
@get("/1")
async def controller(cache: Cache) -> None:
    await cache.get()  # >>> Value


# dependencies.py
# XXX: Dependency Injector can cast container to `DynamicContainer` and Starlite's type check would fail.
ContainerT: TypeAlias = Container | DynamicContainer


async def get_cache(container: ContainerT) -> Cache:
    return await container.cache()


def create_dependencies() -> dict[str, Provide]:
    return {"cache": Provide(get_cache)}


# main.py
async def on_startup(*_, container: Container, **__) -> None:
    # XXX: can't pass `container.init_resources()` directly into `Starlite.on_startup` because of AnyIO worker thread
    await container.init_resources()


def create_app() -> Starlite:
    container = Container()
    container.wire(modules=[__name__])

    def get_container() -> Container:
        return container

    dependencies = {"container": Provide(get_container)}
    dependencies.update(create_dependencies())
    app = Starlite(
        route_handlers=[controller],
        dependencies=dependencies,
        on_startup=[partial(on_startup, container=container)],
    )
    return app

[peterschutt]

How about something like this?

from typing import AsyncIterator

from dependency_injector import containers, providers

from starlite import Provide, Starlite, State, get


# cache.py
class Cache:
    @classmethod
    async def from_url(cls) -> "Cache":
        return cls()

    async def get(self) -> None:
        print("Value")


async def init_cache() -> AsyncIterator[Cache]:
    cache = await Cache.from_url()
    yield cache


# containers.py
class Container(containers.DeclarativeContainer):
    cache = providers.Resource(init_cache)


# controllers.py
@get("/1")
async def controller(cache: Cache) -> None:
    await cache.get()  # >>> Value


# main.py
async def get_cache(state: State) -> Cache:
    return await state.container.cache()


async def on_startup(state: State) -> None:
    # XXX: can't pass `container.init_resources()` directly into `Starlite.on_startup` because of AnyIO worker thread
    container = Container()
    container.wire(modules=[__name__])
    await container.init_resources()
    state.container = container


def create_app() -> Starlite:
    app = Starlite(
        route_handlers=[controller],
        dependencies={"cache": Provide(get_cache)},
        on_startup=[on_startup],
    )
    return app

[ReznikovRoman]

Yes, with State it looks better than my example!

[ThirVondukr]

@ReznikovRoman Seems like you could create your own inject decorator that would use dependency_injector.inject inside and strip parameters that default to Provide[...] so starlite does not complain about them. You can do that by changing function signature:
https://gitlab.com/ThirVondukr/aioinject/-/blob/main/aioinject/utils.py#L8

[ReznikovRoman]

@ThirVondukr Thanks a lot! I finally got dependency-injector working with Starlite!

Here is a complete example for others:

import inspect
import typing
from typing import Annotated, Any, Callable, TypeVar

from dependency_injector import containers, providers, wiring

from starlite import Controller, Starlite, State, get

# lib.dependency_injector
F = TypeVar("F", bound=Callable[..., Any])


def _is_dependency(arg: Any) -> bool:
    return getattr(arg, "__IS_MARKER__", False)


def get_inject_annotations(function: F) -> dict[str, Any]:
    return {
        name: annotation
        for name, annotation in typing.get_type_hints(function, include_extras=True).items()
        if any(_is_dependency(arg) for arg in typing.get_args(annotation))
    }


def clear_wrapper(wrapper: F) -> F:
    inject_annotations = get_inject_annotations(wrapper)
    signature = inspect.signature(wrapper)
    new_params = tuple(
        parameter
        for parameter in signature.parameters.values()
        if parameter.name not in inject_annotations
    )
    wrapper.__signature__ = signature.replace(parameters=new_params)
    for name in inject_annotations:
        del wrapper.__annotations__[name]
    return wrapper


def inject(function: F) -> F:
    wrapper = wiring.inject(function)
    wrapper = clear_wrapper(wrapper)
    return wrapper


# app
class Cache:
    async def get(self) -> str:
        return "value"


class Container(containers.DeclarativeContainer):
    cache = providers.Factory(Cache)


class Resource(Controller):

    @get("/1")
    @inject
    async def controller(
        self,
        cache: Annotated[Cache, wiring.Provide] = wiring.Provide[Container.cache],
    ) -> dict[str, str]:
        return {"cache": await cache.get()}


async def on_startup(state: State) -> None:
    container = Container()
    container.wire(modules=[__name__])
    state.container = container


def create_app() -> Starlite:
    app = Starlite(route_handlers=[Resource], on_startup=[on_startup])
    return app

[peterschutt]

Adding some detail from a discussion with @ThirVondukr in discord.

@ThirVondukr expressed that a weakness with having a built-in DI is the inability to share that DI mechanism with interfaces unrelated to the framework and gave the following example code:

async def get_user(
    token: str = Header(...), # Handled by framework
    user_service: Annotated[UserService, Inject], # Handled by your DI library
) -> User:
    ...

async def endpoint(
  service: Annotated[SomeOtherService, Inject],
  user: User = Depends(get_user),
) -> ...

Example DI frameworks:

• https://python-dependency-injector.ets-labs.org/
• https://github.com/adriangb/di
• https://gitlab.com/ThirVondukr/aioinject/-/blob/main/aioinject/utils.py#L8

Examples of using aioinject:

• fastapi: https://gitlab.com/manga-microservices/identity/-/blob/main/src/apps/auth/router.py#L24
• strawberry: https://gitlab.com/manga-microservices/identity/-/blob/main/src/gql/users/mutation.py#L19

Would certainly be interesting to investigate whether we could facilitate our DI to interop with framework DI in some manner.

[ThirVondukr]

@peterschutt Essentially modifying wrapped function is up to DI framework or people who use it, I don't think that starlite could provide something for that case, all DI libraries use something different.

For example:

• di uses a mix of bare annotations and Annotated[... Marker] in entrypoints.
• Dependency Injector seems to be using parameter default values to store it's markers.
• In aioinject on entrypoint level I only used Annotated[..., Inject] so it's easy to spot a dependency and work with it programmatically.

[peterschutt]

It very well may be the case that there is nothing to do here.

As always, thanks for input.

[ThirVondukr]

Also regarding interepability - di and aioinject should work with following code:

@inject
async def dependency(
    request: Request,
    user_service: Annotated[UserService, Inject],
) -> User:
    ...
    
@get("/", dependencies={"user": Provide(get_user)})
@inject
async def endpoint(
    user: User,
    another_service: Annotated[AnotherService, Inject],
): 
    ...

However if I remember correctly dependency-injector does not have a well defined context and you might receive different instances of your services here, which would not be desirable in a lot of cases, for example when using sqlalchemy session/connection

So, yep, it seems like nothing has to be done on starlite side

[ReznikovRoman]

Yep, I can confirm that Starlite works fine with aioinject

Here is a complete example if someone is interested:

from typing import Annotated, Any, Callable, TypeVar

import aioinject
from aioinject import Container, Inject, providers, utils
from aioinject.decorators import InjectMethod

from starlite import Controller, Starlite, State, get
from starlite.types import ASGIApp, Receive, Scope, Send

# aioinject.ext.starlite
F = TypeVar("F", bound=Callable[..., Any])


def inject(function: F) -> F:
    wrapper = aioinject.decorators.inject(function, inject_method=InjectMethod.context)
    wrapper = utils.clear_wrapper(wrapper)
    return wrapper


def inject_middleware(app: ASGIApp) -> ASGIApp:
    async def middleware(scope: Scope, receive: Receive, send: Send) -> None:
        async with scope["app"].state.container.context():
            await app(scope, receive, send)
    return middleware


async def on_startup(state: State) -> None:
    container = Container()
    container.register(providers.Callable(Cache))
    state.container = container


# app
class Cache:
    async def get(self) -> None:
        print("Value")


class Resource(Controller):

    @get("/1")
    @inject
    async def controller(self, cache: Annotated[Cache, Inject]) -> dict[str, str]:
        await cache.get()
        return {"hello": "world"}


def create_app() -> Starlite:
    app = Starlite(route_handlers=[Resource], on_startup=[on_startup], middleware=[inject_middleware])
    return app

@ThirVondukr I think you can even add a Starlite plugin for your DI framework

[peterschutt]

Nice!

[ThirVondukr]

@ReznikovRoman I absolutely missed that mention 😅
I just added litestar integration via plugin: https://gitlab.com/ThirVondukr/aioinject/-/merge_requests/3