Dependency injection

[filwaline]

Any suggestions on how to use dependency injection with Pyventus?

For example, for a database connection, how can I pass this dependency to a Pyventus event handler?

I have found two methods.

One is to wrap the handler as a class method so that it can access instance attributes that are injected during class initialization.

from pyventus import EventEmitter, EventLinker, AsyncIOEventEmitter
from dataclasses import dataclass


@dataclass
class DemoEvent:
    a: str
    b: int


class DemoHandler:
    def __init__(self, deps):
        self.deps = deps

    def __call__(self, event: DemoEvent, extra_args):
        print("DemoHandler called with args:", event)
        print("Dependencies:", self.deps)
        print("Dependencies from extra args:", extra_args.get("deps_emit"))
        return


emitter: EventEmitter = AsyncIOEventEmitter()
EventLinker.subscribe(
    DemoEvent,
    event_callback=DemoHandler(deps={"uow": "unit of work"})
)

emitter.emit(DemoEvent(a="Hello", b=1), extra_args={"deps_emit": "ddd"})

The second way is to pass the dependency as an extra parameter when emitting an event. However, the second method seems not to work if we are using some processor that requires event parameters to be serializable.

What was your intention? Or is there some support for dependency injection that I have missed?

[mdapena]

Hi @filwaline,

Thanks for bringing up this interesting question and use case for Pyventus!

Current State

By default, Pyventus doesn’t include a dependency injection system or utilities for this purpose. This is because the EventLinker is solely designed to orchestrate the linkage of events and their inherent logic (callbacks), while the EventEmitter manages the event emission and its execution, using the current context and the given payload. As a result, event handlers will have access to what is available in the current callable context, the event payload, or the callable object, similar to what a callback will have when it gets executed.

Considerations

There are multiple ways to tackle this problem, but it’s important to keep in mind that dependencies shouldn’t be passed through events, as they should only contain relevant information about the event itself. This is not only a matter of good practice but also crucial when the event needs to be serialized or deserialized for processing, which can be tricky and prone to errors. Additionally, events should not be tightly coupled with implementation details, such as how they will be processed.

Proposed Solution

Now, one way to solve this problem, considering all of the above and in a non-tedious manner, is through the use of decorators. Since event handlers are just functions, we can wrap them using decorators to add any additional behavior, such as injecting dependencies.

Let’s first define the main decorator that, when the event handler is called, will create an instance of the given dependency using a factory method style and inject it into the event callback as the first positional argument, followed by the rest.

def depends(factory: Callable | Generator):
    """Decorator for handling dependency injection."""

    def decorator(event_handler):
        """Wraps the event handler (the function subscribed to the event)."""

        @wraps(event_handler)
        def wrapper(*args, **kwargs):
            """Executes the event handler with the provided dependency."""

            # Call the original event handler with the dependency
            if isgeneratorfunction(factory):
                gen = factory()
                try:
                    dependency = next(gen)
                    return event_handler(dependency, *args, **kwargs)
                finally:
                    gen.close()
            else:
                dependency = factory()
                return event_handler(dependency, *args, **kwargs)

        return wrapper

    return decorator

Of course, this can be further improved (e.g., with typing autocompletion, support for multiple dependencies, etc.), but it serves as a good starting point. With this decorator, let’s create an example in which an event is emitted, and the event handler requires a database connection to process the event.

@dataclass
class DBSession:
    id: int


def get_db_session() -> Generator[DBSession, None, None]:
    """Generator that emulates a factory for database sessions."""

    try:
        print("Starting database session")
        yield DBSession(id=random.randint(100000, 999999))
    finally:
        print("Database session closed.\n")


@dataclass
class DemoEvent:
    a: str
    b: int


@EventLinker.on(DemoEvent)
@depends(get_db_session)  # Inject the database session factory as a dependency when the event handler is executed
def event_handler_with_dependency_injection(session: DBSession, event: DemoEvent) -> None:
    """Event handler that uses dependency injection to obtain a database session each time it is executed."""

    print(f"\tDependency: {session}")
    print(f"\tEvent: {event}")


if __name__ == "__main__":
    # Using a ProcessPoolExecutor to emulate different processes/context/serialization
    with ExecutorEventEmitter(ProcessPoolExecutor()) as event_emitter1:
        event_emitter1.emit(event=DemoEvent(a="First Event", b=1))

    # Using an AsyncIOEventEmitter to emulate the same process/context/no serialization
    event_emitter2 = AsyncIOEventEmitter()
    event_emitter2.emit(event=DemoEvent(a="Second Event", b=2))

Click here to view the full example...

from collections.abc import Callable
from concurrent.futures import ProcessPoolExecutor
from dataclasses import dataclass
from functools import wraps
from inspect import isgeneratorfunction
import random
from typing import Generator

from pyventus import AsyncIOEventEmitter, EventLinker, ExecutorEventEmitter


def depends(factory: Callable | Generator):
    """Decorator for handling dependency injection."""

    def decorator(event_handler):
        """Wraps the event handler (the function subscribed to the event)."""

        @wraps(event_handler)
        def wrapper(*args, **kwargs):
            """Executes the event handler with the provided dependency."""

            # Call the original event handler with the dependency
            if isgeneratorfunction(factory):
                gen = factory()
                try:
                    dependency = next(gen)
                    return event_handler(dependency, *args, **kwargs)
                finally:
                    gen.close()
            else:
                dependency = factory()
                return event_handler(dependency, *args, **kwargs)

        return wrapper

    return decorator


@dataclass
class DBSession:
    id: int


def get_db_session() -> Generator[DBSession, None, None]:
    """Generator that emulates a factory for database sessions."""

    try:
        print("Starting database session")
        yield DBSession(id=random.randint(100000, 999999))
    finally:
        print("Database session closed.\n")


@dataclass
class DemoEvent:
    a: str
    b: int


@EventLinker.on(DemoEvent)
@depends(get_db_session)  # Inject the database session factory as a dependency when the event handler is executed
def event_handler_with_dependency_injection(session: DBSession, event: DemoEvent) -> None:
    """Event handler that uses dependency injection to obtain a database session each time it is executed."""

    print(f"\tDependency: {session}")
    print(f"\tEvent: {event}")


if __name__ == "__main__":
    # Using a ProcessPoolExecutor to emulate different processes/context/serialization
    with ExecutorEventEmitter(ProcessPoolExecutor()) as event_emitter1:
        event_emitter1.emit(event=DemoEvent(a="First Event", b=1))

    # Using an AsyncIOEventEmitter to emulate the same process/context/no serialization
    event_emitter2 = AsyncIOEventEmitter()
    event_emitter2.emit(event=DemoEvent(a="Second Event", b=2))

As shown in the code example, this function works under all conditions, whether in the same context or in another process, because the only thing that is serialized and deserialized is the event itself, nothing else.

With that being said, I hope this example is useful for your particular use case and for everyone else as well. Again, this is not the final solution; there are multiple approaches, so if anyone else wants to contribute, they are more than welcome to do so.

Happy coding!

[filwaline]

I have found a dependency injection package that meets my requirements. This package allows me to declare and initialize dependencies in a location other than an event handler's decorator. As a result, I can modify the dependencies without having to modify the parts of the event handler.

dependency-injector https://python-dependency-injector.ets-labs.org/index.html

example code:

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from dataclasses import dataclass
import random
from pyventus import AsyncIOEventEmitter, EventLinker, ExecutorEventEmitter
from dependency_injector import containers, providers
from dependency_injector.wiring import Provide, inject


@dataclass
class DBSession:
    id: int


class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    gen = providers.Singleton(lambda: random.randint(100000, 999999))

    session = providers.Factory(DBSession, id=gen)


@dataclass
class DemoEvent:
    a: str
    b: int


@EventLinker.on(DemoEvent)
@inject
def event_handler_with_dependency_injection(
    event: DemoEvent,
    session: DBSession = Provide[Container.session],
) -> None:
    """Event handler that uses dependency injection to obtain a database session each time it is executed."""

    print(f"\tDependency: {session}")
    print(f"\tEvent: {event}")


def init():
    container = Container()
    container.init_resources()
    container.wire(modules=[__name__])
    return container


if __name__ == "__main__":
    # Using a ProcessPoolExecutor to emulate different processes/context/serialization
    with ExecutorEventEmitter(ProcessPoolExecutor(initializer=init)) as event_emitter1:
        event_emitter1.emit(event=DemoEvent(a="First Event", b=1))

    # Using an AsyncIOEventEmitter to emulate the same process/context/no serialization
    container = init()
    event_emitter2 = AsyncIOEventEmitter()
    event_emitter2.emit(event=DemoEvent(a="Second Event", b=3))

    with container.override_providers(gen=providers.Singleton(lambda: 99999999)):
        with ExecutorEventEmitter(ThreadPoolExecutor()) as event_emitter3:
            event_emitter3.emit(event=DemoEvent(a="Third Event", b=5))

[mdapena]

Awesome! That looks great. Using a package like that really simplifies things and adds flexibility.

Thanks for sharing!