reagento / dishka Goto Github PK

Add option to use lock in container/subcontainer

Sometimes we have a dependency provided by library and we want to modify it (like decorating).
So, the next provider will receive some dependency (+additional), and provide the same type of dependency

Do you happen to have a fastapi sqlalchemy async example to use this library with?

When we finalize dependencies we have no information if there were exception during process handling.

We can use send or asend to pass it without breaking compatibility.

Also we will need to add an optional parameter to close method if Container object and modify it's __exit__

Sometimes (especially when we follow interface segregation principle) we have single implementation for multiple dependencies. It would be useful to have single instance for all of them.

Let's
Current way to make alias:

class MyProvider(Provider):
    @provide(scope=MyScope.REQUEST)
    def get_repo(self, repo: UserRepositoryImpl) -> UserGetter:
       return repo

We can rely on the fact that scope must be the same, so this can be simplified to:

class MyProvider(Provider):
    repo = alias(UserRepositoryImpl, dependency=UserGetter)

Create something depending on, at least, request data

Add tests and ensure that everything is working as intended.

Allow to set alias when entering scope. That will allow to change logic for specific route without moving it to container

with container(aliases={A: A1}) as request_container:
   request_container.get(A) # this will reuturn A1

Aliases should be copied to inner scopes if we do not know the scope of depedency. But the do not affect the outer scope.

Add function to render all dependencies into one file. HTML with Mermaid can be good for start

Do not allow to override dependency provider unless it is explicitly declared

E.g. fastapi app and bot in one process will have single app container

Sometimes we want to have additional itnermediate scopes in some run configurations. E.g

1. In tests we want finalize app dependencies while keeping some objects in cache. In can be solved adding RUNTIME scope before APP
2. For Websocket we have additional scope for connected client which doesn't exist for HTTP requests. Though, many object will be of the REQUEST scope.

The idea consists of two parts:

1. An option to specificy scope when entering context. With this you will enter multiple scopes at one call. Exit will also rewind all entered scopes.
2. An option on scope iteself to be skipped when entering wihout explicit scope.

So, if you enter context you
a) either provide scope explicitely, then all intermediate scopes will be also entered
b) do not provide any scope, then it is treated as the next non-skippable scopes. All scopes with mark skip will be enterd automatically

Hey!

Is it possible override dependecies like it realized in dependency-injector?

with container.api_client_factory.override(unittest.mock.Mock(ApiClient)):
    service2 = container.service_factory()
    assert isinstance(service2.api_client, unittest.mock.Mock)

if I've forgot scope i received this error

E               KeyError: None

..\..\AppData\Local\pypoetry\Cache\virtualenvs\shvatka-7O9jFSMq-py3.11\Lib\site-packages\dishka\registry.py:84: KeyError

please add to error provider name (and will be greate if method name or provided type for case with really many @provide functiuons in one provider)

if I created sync container and add async provider i receive next error:

ValueError: Unsupported type FactoryType.ASYNC_GENERATOR

will be greate to add to error why (wrong type of container)

If my object supports few interfaces I need to write wrapper for each one.
Can you support some syntaxes for single object which support several interfaces?
For example:

@provide(scope=Scope.REQUEST) 
def get_db(self) -> I1 | I2 | I3:
    return object()

Дальше видно, что не-optional аргумент проверяется на None.

How can I use injection on classes which are initialised on each route? The following code results in this error which is logical because I use FastAPI Depends. The error:

    raise fastapi.exceptions.FastAPIError(
fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'common.services.permission_service.PermissionService'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None. Read more: https://fastapi.tiangolo.com/tutorial/response-model/

My auth route:

@router.post('/token', response_model=BaseResponse[GetAuthTokenResponse])
@inject
async def token_authorize(request_data: AuthTokenRequest, user: AnnotatedAuthenticatedUser, auth_service: Annotated[AuthService, FromDishka()]):
    user, token = ....
    return schemas.BaseResponse[GetAuthTokenResponse](data={
        'user': user,
        'token': token.decode(),
    })

the AnnotatedAuthenticatedUser:

AnnotatedAuthenticatedUser = Annotated[AuthenticatedUser, Depends(AuthCheck())]

And the AuthCheck:

class AuthCheck:
    def __init__(self, permission: AuthCheckPermission = None):
        self._permission = permission

    def __call__(self,
                res: Response,
                req: Request,
                permission_service: Annotated[PermissionService, FromDishka()],
                invite_service: Annotated[InviteService, FromDishka()],
                event_service: Annotated[event_service.EventService, FromDishka()],
                organisation_service: Annotated[OrganizationService, FromDishka()],
                cred: HTTPAuthorizationCredentials=Depends(HTTPBearer(auto_error=False)),
    ):
        # with trace.get_tracer(__name__).start_as_current_span("auth_check") as _:
        if cred is None:
            raise InvalidAuthException
        try:
            decoded_token = auth.verify_id_token(cred.credentials)
        except Exception as err:
            raise InvalidAuthCredentialsException(original_exception=err)
...

Let's image we have 3 classes:

class C(Generic[T]):
    pass

class Gw(Generic[T]):
    def __init__(self, c: C[T]): ...

class U:
    def __init__(self, Gw[Model]): ...

We should not declare factories for concrete intermediate classes like Gw in example. E.g. this is expected to work:

class MyProvider(Provider):
    @provide(scope=Scope.APP)
    def c_model(self) -> C[Model]: 
       return CModelImpl()

    gw = provide(Gw, scope=Scope.APP)
    u = provide(U, scope=Scope.APP)
...

container.get(U)

Add LazyProxy and put it instead of requested class when cycle is detected

I'm moving from the python-dependency-injector library to this one and am using a service-repo pattern.
This is my small setup:

class DBProvider(Provider):
    @provide(scope=Scope.APP)
    def engine(self) -> AsyncEngine:
        engine = create_async_engine(
            get_config().ASYNC_DB_URL(),
            echo=True,
        )
        return engine


    @provide(scope=Scope.REQUEST)    
    async def session(self, engine: AsyncEngine) -> AsyncIterable[AsyncSession]:
        async with AsyncSession(engine) as session:
            yield session



class Interactor:
    def __init__(self, session: AsyncSession):
        self.session = session

    def test(self) -> str:
        return "test"
    

class InteractorProvider(Provider):
    i1 = provide(Interactor, scope=Scope.REQUEST)


class ServiceProvider(Provider):
    wallet_pass_api_service = provide(WalletPassApiService, scope=Scope.APP)
    block_user_repo = provide(BlockedUserRepository, scope=Scope.REQUEST)


container = make_async_container(
    DBProvider(), 
    ServiceProvider(),
    InteractorProvider()
)

When I use Scope.APP with the block_user_repo this does not work, as expected, since the BlockedUserRepositoryhas a dependency on theAsyncSessionwhich is aScope.REQUEST`.

But shouldn't all repositories and service be initialised on app startup? Because creating, possibly 10 or 20 repositories/services on each request result in large overhead?

it is exhausting to declare scope for each depedency as most of the will have Request scope. It woul be easier to set it on Provder level. E.g

class MyProvider(Provider):
    scope = Scope.REQUEST

    a = provide(A)   # will have Scope.REQUEST
    b = provide(B, scope=Scope.APP)   # will have Scope.APP

We need to ensure that all required dependencies can be actually created on startup.

• all nodes of graph are reachable #85
• all context varibles are passed
• all requested types are reachable

E.g.

with container(request=value) as subcontainer:
   ...

request here should be available in solving context.
Or using typehint due to we don't have named dependencies:

with container({Request: value}) as subcontainer:
   ...

special marker to provide element of a list or mapping

Idea:
I suggest adding the ability to disable lazy dependency loading for the 'provide' function or globally for the entire 'Provider'

How is it supposed to work?
When creating a new container, create all dependencies with the 'lazy=False' flag and add them to the cache.

Why do you need it?
Create resource-intensive sessions or pools, as well as detect errors early when creating dependencies and protect against errors during code execution

How I see it in the code
Provide

class MyProvider(Provider)
    a = provide(A, scope=Scope.App, lazy=False)

Provider

class MyProvider(Provider)
    lazy = Fasle
    a = provide(A, scope=Scope.App)

• Intro into dependency injection
• Intro into IoC-containers
• why we need scopes
• what is providers
• how to use container
• async vs sync
• decorator
• alias
• Integrations:
  • Flask
  • Fastapi
  • Litestar
  • aiogram
  • pytelegrambotapi
• comparation with other libraries
  • rodi
  • di
  • dependency injector
  • fastapi depends

Use case:
Interactor depends on StatisticDbGateway and MainDbGateway. Each of them uses SQLAlchemy Session object, so they have similar init typehints. The difference is that StatisticDbGateway requires Session bound to Clickhouse Engine, while MainDbGateway binds to postgresql Engine. Again both engines have same type, but created differently.

To solve this you can replace Session with some new type (literally NewType("PgSession", Session)), but it requires modification of all code base. Imagine that you took those objects from external library and cannot modify it.

Another solution is again to use NewType, but create factories for all objects instead of registrering classes as depedencies directly. This requries a lot of work.

Proposed solution is to split those objects in 3 groups:

SubGraph1: StatisticDbGateway, Session, Engine
SubGraph2: MainDbGateway, Session, Engine
MainGraph: Interactor, SubGraph1, SubGraph2

Each graph here uses only his factories to solve dependencies or requests specially attached subgraphs. So here, gateways can request Session and they get exactly that session which is declared in his subgraph.

Requirements:

1. (Sub)graphs can have subgraphs
2. All (Sub)graphs are synchronized by scopes
3. Subgraph cannot access depedencies from parent graph
4. When looking for factories subgraphs are requested only if they are not found in current graph itself.

It is an open question whether (sub)graph is just a container or another thing

Hi,

How could one use dishka with FastStream? https://github.com/airtai/faststream