Is generic provider/container possible to implement?

[dartt0n]

Is it possible to create generic container or provider?

Basically, the idea is to have multiple generic interfaces, and require a specific implementation to specify the type. Then, I want to have a dependency on these interfaces in the usecase class (in the example below: to require all interface implementations to use exactly the same session type).

class MyUsecase[UserSession, AnotherSession]:
    def __init__(
        self,
        adapter1: UserRepoInterface[UserSession],
        adapter2: TokenRepoInterface[UserSession], # these two are restricted to use the same session
        adapter3: LogRepoInterface[AnotherSession], # this one is free
    ):
        ...

This can be solved by having the interface accept/return the session interface rather than the bounded generic type, but then I have no way to make sure that all adapters use the same session.

Some draft of the idea:

import random
import string
from dataclasses import dataclass
from typing import Protocol, final, override

from dishka import Provider, Scope, make_container, provide


class SessionInterface(Protocol):
    def begin(self) -> None: ...
    def savepoint(self) -> str: ...
    def commit(self) -> None: ...
    def rollback(self, savepoint: str | None = None) -> None: ...


class RepoInteface[SessionT: SessionInterface](Protocol):
    def get_session(self) -> SessionT: ...
    def save_user(self, session: SessionT, name: str) -> None: ...


@dataclass
class Usecase[T: SessionInterface]:
    repo: RepoInteface[T]

    def do_business_logic(self):
        session = self.repo.get_session()

        self.repo.save_user(session, "Ivan")


@dataclass
class MySessionImpl:
    name: str

    def begin(self) -> None:
        print(f"begin; -- {self.name}")

    def savepoint(self) -> str:
        random_string = "".join(random.choices(string.ascii_lowercase, k=10))
        print(f"savepoint {random_string}; -- {self.name}")
        return random_string

    def commit(self) -> None:
        print(f"begin; -- {self.name}")

    def rollback(self, savepoint: str | None = None) -> None:
        print(f"rollback{' ' + savepoint if savepoint else ''}; -- {self.name}")


@dataclass
class RepoImpl(RepoInteface[MySessionImpl]):
    @override
    def get_session(self) -> MySessionImpl:
        return MySessionImpl("session_my_repo_impl")

    @override
    def save_user(self, session: MySessionImpl, name: str) -> None:
        session = self.get_session()
        session.begin()

        print(f"INSERT INTO users (name) VALUES ({name:!r});")  # dont do this in production

        session.commit()


@final
class MyProvider[T: SessionInterface](Provider):
    # generic provider
    scope = Scope.APP

    usecase = provide(source=Usecase[T])


# in prod
container = make_container(MyProvider[MySessionImpl]())
# in tests: container = make_container(MyProvider[FakeSession])

create_user_usecase = container.get(Usecase)
create_user_usecase.do_business_logic()

[Tishka17]

You do not need the whole provider to be genetic. Just create generic method, it should work with provide. See doc for examples

[dartt0n]

The problem is that the current implementation on the generic method provide requires that for all generic dependencies, a generic provide factory must be defined. So, if I have 10 generic usecases, I must provide 10 generic factories for usecases + N generic factories for all the generic dependencies of these usecases.

That is, currently it's impossible to have two different providers for the application for different contexts:

class RealProvider(Provider):
    scope = Scope.APP
    real_repo = provide(source=RealRepo, provides=RepoInteface[RealSession])

class FakeProvider(Provider):
    scope = Scope.APP
    fake_repo = provide(source=FakeRepo, provides=RepoInteface[FakeSession])

class UsecasesProvider(Provider):
    scope = Scope.APP

    @provide
    def create_user_usecase[T: SessionInterface](
        self, type_guard: type[T], repo: RepoInteface[T]
    ) -> CreateUserUsecase[T]:
        _ = type_guard
        return CreateUserUsecase[T](repo)


# in main.py
container = make_container(UsecasesProvider(), RealProvider())
create_user_usecase = container.get(CreateUserUsecase[RealSession])  # NoFactoryError
create_user_usecase.do_business_logic()

# in test_some_logic.py
container = make_container(UsecasesProvider(), FakeProvider())
create_user_usecase = container.get(CreateUserUsecase[FakeSession])  # NoFactoryError
create_user_usecase.do_business_logic()

Such code fails with an error that there is no provider for generic RepoInterface[T].

This can be fixed by adding a generic factory for this interface, for example like this:

# ... 

class UsecasesProvider(Provider):
    scope = Scope.APP

    @provide
    def repo[T: SessionInterface](
        self,
        type_guard: type[T],
        real_repo: RepoInteface[RealSession],
        fake_repo: RepoInteface[FakeSession],
    ) -> RepoInteface[T]:
        if type_guard is RealSession:
            return real_repo

        if type_guard is FakeSession:
            return fake_repo

        msg = f"Unsupported session type: {type_guard}"
        raise ValueError(msg)

# ...

container = make_container(UsecasesProvider(), RealProvider(), FakeProvider())
create_user_usecase = container.get(CreateUserUsecase[RealSession])  # works
create_user_usecase.do_business_logic()

But then, the following problems appear:

1. Now the usecase provider must know about ALL possible variants of generic T
2. An instance of each repository variant must be instantiated (here: RealRepo & FakeRepo)
3. When adding a new variant of T, we must:
   • Add new provider
   • Update generic repo provider
   • Add new provider to container EVERYWHERE
4. And more, now it's possible to create Usecase[FakeSession] in "production" enviroment, although this should be prohibited

So basically, with these problems, it makes no sense to use generics at all, because the way of using them contradicts the very idea of generics.

Here, I expect generic provide factory to work similar to how summon and given works in scala:

given realRepo: RepoInterface[RealSession] = RealRepo()
given fakeRepo: RepoInterface[FakeSession] = FakeRepo()

object UsecaseFactory:
  def createUserUsecase[T <: SessionInterface](using RepoInterface[T]): CreateUserUsecase[T] =
    CreateUserUsecase[T](summon[RepoInterface[T]])


@main def summonGenericUsecase(): Unit =
  println("=== Running with Real implementations ===")
  val realUsecase = UsecaseFactory.createUserUsecase[RealSession]
  realUsecase.doBusinessLogic()

  println("=== Running with Fake implementations ===")
  val fakeUsecase = UsecaseFactory.createUserUsecase[FakeSession]
  fakeUsecase.doBusinessLogic()

[full scala snippet]

[Tishka17]

Let's step to the beginning of the thread.

class MyUsecase[UserSession, AnotherSession]:
    def __init__(
        self,
        adapter1: UserRepoInterface[UserSession],
        adapter2: TokenRepoInterface[UserSession], # these two are restricted to use the same session
        adapter3: LogRepoInterface[AnotherSession], # this one is free
    ):
        ...

Is it really generic? I see that you call it as create_user_usecase = container.get(Usecase), so it not generic. Repos can be still generics.

What I suggest here:

UserSession = NewType("UserSession", SessionInterface)
AnotherSession = NewType("UserSession", SessionInterface)


class MyUsecase:  # not generic 
    def __init__(
        self,
        adapter1: UserRepoInterface[UserSession],  # specialised generics
        adapter2: TokenRepoInterface[UserSession], 
        adapter3: LogRepoInterface[AnotherSession],
    ):
        ...


T = TypeVar("T")

class MyProvider(provider):
   usecase = provide(MyUsecase)  # not generic, depends on specialized repos
   user_repo = provide(UserRepoImpl[T], provides=UserRepoInterface[T])  # generic factory
   token_repo = provide(TokenRepoImpl[T], provides=TokenRepoInterface[T])
   log_repo = provide(LogRepoImpl[T], provides=LogRepoInterface[T])
   
   @provide
    def user_session(self) -> UserSession:
        return ...

   @provide
    def user_session(self) -> AnotherSession:
        return ... 

Also, you do not need all this to be a single provider, you can create multiple classes to be able to combine them

[Tishka17]

Another solution is to use components and no generics:

class RepoProvider(Provider):
   user_repo = provide(UserRepoImpl, provides=UserRepoInterface)
   token_repo = provide(TokenRepoImpl, provides=TokenRepoInterface)
   log_repo = provide(LogRepoImpl, provides=LogRepoInterface)

   @provide
   def session(self) -> Session:
        return ...


USER_COMPONENT = "UserComponent"
ANOTHER_COMPONENT = "AnotherComponent"

class UseCaseProvider(Provider)
    @provide
    def user_case(
         self,
        user_repo: Annotated(UserRepoInterface, FromComponent(USER_COMPONENT)),
        token_repo: Annotated(TokenRepoInterface, FromComponent(USER_COMPONENT)),
        log_repo: Annotated(LogRepoInterface, FromComponent(ANOTHER_COMPONENT)),
    ):
       return UseCase(user_repo, token_repo, log_repo)

container = make_container(
     UseCaseProvider(),
     RepoProvider().to_component(USER_COMPONENT),
     RepoProvider().to_component(ANOTHER_COMPONENT),
)

[Tishka17]

If the only idea was to have separate configuration for production and testing: just create 2 providers for repos. One for production, second one for testings. They will provide the same types, but create different implementations. Later, when building a container select which on to use:

if config.producton:
   repos = ProductionRepoProvider()
else:
   repos = TestingRepoProvider()

container = make_container(UsecaseProvider(), repos)

Also, take a look on new when= feature: https://dishka.readthedocs.io/en/stable/advanced/when.html

[dartt0n]

Regarding whether a usecase should be generic or not. This is quite a philosophical question, but I think this way: "An usecase should be an implementation of business logic and should have zero knowledge and dependencies on specific technologies and/or implementations". In my case, there is an usecase that works with three different adapters, while I want to restrict two of the adapters to strictly one type of session/context (specific objects are not important here, but the point is that through generics, we can pervasively specify the dependency of these adapters on one technology, while not binding and combining them into 1 large adapter).

Unfortunately, NewType is not very useful here, because NewType is not contravariant. See:

SessionType = NewType("SessionType", SessionInterface)

class Usecase:
    def __init__(self, repo: RepoInteface[SessionType]): ...

class MyProvider(Provider):
    scope = Scope.APP
    
    # Here we specify RepoInteface[MySessionImpl] because
    # adapter indeed implements RepoInteface[MySessionImpl]
    # adapter does not know anything about SessionType from usecase
    repo = provide(source=RepoImpl, provides=RepoInteface[MySessionImpl])
    
    ...
    usecase = provide(source=Usecase)


container = make_container(MyProvider()) # GraphMissingFactoryError
# this happens because MySessionImpl is not subtype of SessionType.
# MySessionImpl is subtype of SessionInterface.
# And since SessionType is not contravariant
#  the property MySessionImpl <: SessionType does not hold

create_user_usecase = container.get(Usecase)
create_user_usecase.do_business_logic()

However, we can change the code like this:

-    repo = provide(source=RepoImpl, provides=RepoInteface[MySessionImpl])
+    repo = provide(source=RepoImpl, provides=RepoInteface[SessionType])

But in this case, this code should now be located next to the usecase, because SessionType is in the usecase's ownership. Accordingly, it turns out that you need to directly specify which specific adapter needs to be used for the usecase (e.g. by specified one session type over another). And this already loses any DI meaning in such a scenario, since now it is no different from directly passing values to the usecase's constructor.

I am now diving deeper into @dishka.decorate, and it seems to me that with this, it will be possible to implement an analogue of Scala's given/summon. If it works out - I'll publish the solution in this issue.

In any case, thank you very much for the suggested alternatives, I'm sure they will help many users, but unfortunately they don't solve my problem.

[Tishka17]

I do not get why did you wrote specialized factory, providing only single type

repo = provide(source=RepoImpl, provides=RepoInteface[MySessionImpl])

while I suggested using generic factory like

repo = provide(source=RepoImpl[T], provides=RepoInteface[T])

[dartt0n]

I do not get why did you wrote specialized factory, providing only single type

repo = provide(source=RepoImpl, provides=RepoInteface[MySessionImpl])
while I suggested using generic factory like

repo = provide(source=RepoImpl[T], provides=RepoInteface[T])

RepoImpl is not a generic class, it is a concrete implementation for RepoInteface[SomeType].
For example:

class UserRepo[SessionT](Protocol):
    @asynccontextmanager
    async def context(self) -> AsyncGenerator[SessionT]: ...

    async def create_user(self, session: SessionT, name: str) -> domain.User: ...

class PostgresUserRepoAdapter(UserRepo[sqlalchemy.AsyncSession]):
    @asynccontextmanager
    async def context(self) -> AsyncGenerator[sqlalchemy.AsyncSession]:
        async with self.__session_factory() as session:
            yield session
            
    async def create_user(self, session: sqlalchemy.AsyncSession, name: str) -> domain.User:
        user = orm.User(name=name)
        session.add(user)
        session.flush()

        return user.to_domain()

[dartt0n]

I have found a workaround.

You can generate factories in a runtime using simple container factory:

def build_container[T](type_guard: type[T]) -> Container:
    usecase_provider = Provider()
    usecase_provider.provide(UseCase[type_guard])
    # note this                      ^^^^^^^^^^

    return make_container(
        provider,
        RealProvider(),
        FakeProvider(),
        InfraProvider(),
    )


usecase_real = build_container(RealSession).get(UseCase[RealSession]) # works

usecase_fake = build_container(FakeSession).get(UseCase[FakeSession]) # works

[dartt0n]

Github repository with simple demo

[Tishka17]

Technically, if you use generics, UseCase is unparametrized and should not be used. Type checkers may treat is as a synonym for UseCase[Any] which is definitely a different type from UseCase[RealSession]. Looks like a bug in dishka, it shouldn't work

In the provided link I see the different .get() param, more expected for me

[dartt0n]

Sorry, that was a typo. Updated the comment as follows:

-usecase_real = build_container(RealSession).get(UseCase) # works
+usecase_real = build_container(RealSession).get(UseCase[RealSession]) # works

-usecase_fake = build_container(FakeSession).get(UseCase) # works
+usecase_fake = build_container(FakeSession).get(UseCase[FakeSession]) # works

[dartt0n]

From a more technical side, key issue is a Python generics design:

class MyProvider[T](Provider):
    #           ^^^ this is TypeVar with context = class MyProvider

    #                               ... this is TypeVar with context = class MyProvider
    usecase = dishka.provide(UseCase[T])
    #                ^^^^^^^
    #                however, this executes BEFORE finishing class MyProvider context
    #                therefore, T because bounded as TypeVar, not concrete type.
    #                So, when MyProvider[Session] is created,
    #                MyProvider[Session].usecase still depends on T@MyProvider

However in runtime, only concrete types exist and can be passed as values:

def build_container[T](type_guard: type[T]) -> Container:
    #              ^^^ this is TypeVar: T@build_container
    
    usecase_provider = Provider()
    usecase_provider.provide(UseCase[type_guard])
    #                                ^^^^^^^^^^
    # However, this is a concrete type already known

    return make_container(
        provider,
        RealProvider(),
        FakeProvider(),
        InfraProvider(),
    )

Following the same logic about runtime-known generic concrete types (concrete types are known at a compile time in compiled languages, but python is not one), the following code is also valid:

class MyProvider(dishka.Provider):
    scope = dishka.Scope.APP

    @classmethod
    def guard[T](cls, type_guard: type[T], *args, **kwargs):
        cls.usecase = dishka.provide(UseCase[type_guard])

        return cls(*args, **kwargs)

# works
container = dishka.make_container(MyProvider.guard(RealSession))

# works
usecase_real = .get(UseCase[RealSession])

So, at the end, this issue is a limitation of Python and not related to dishka