Dependency Injection

DependencyProvider() is an interface for any concrete provider capable of resolving and matching function parameters. Both Application() and TransactionContext() internally use dependency provider to resolve handler parameters. By default use dict based implementation BasicDependencyProvider().

This code demonstrates basic functionality of a dependency provider

from lato.dependency_provider import BasicDependencyProvider

class FooService():
    pass

def a_handler(service: FooService):
    pass

foo_service = FooService()
dp = BasicDependencyProvider(foo_service=foo_service)
assert dp[FooService] is foo_service
assert dp["foo_service"] is foo_service

assert dp.resolve_func_params(a_handler) == {'service': foo_service}

lagom integration

This code showcases a dependency provider based on lagom:

import uuid

import lagom.exceptions
from lagom import Container

from lato import Application, DependencyProvider, TransactionContext
from lato.dependency_provider import as_type


class CorrelationId(uuid.UUID):
    pass


class Name(str):
    pass


class Session:
    ...


class Repository:
    def __init__(self, session: Session):
        self.session = session


class Engine:
    def __init__(self, url):
        self.url = url

    def create_sesson(self):
        return Session()


application_container = Container()
application_container[Name] = "Foo"
application_container[Engine] = Engine("sqlite:///:memory:")


class LagomDependencyProvider(DependencyProvider):
    allow_names = False

    def __init__(self, lagom_container):
        self.container = lagom_container

    def has_dependency(self, identifier: str | type) -> bool:
        if type(identifier) is str:
            return False
        return identifier in self.container.defined_types

    def register_dependency(self, identifier, dependency):
        if type(identifier) is str:
            raise ValueError(
                f"Lagom container does not support string identifiers: {identifier}"
            )
        try:
            self.container[identifier] = dependency
        except lagom.exceptions.DuplicateDefinition:
            pass

    def get_dependency(self, identifier):
        if type(identifier) is str:
            raise ValueError(
                f"Lagom container does not support string identifiers: {identifier}"
            )
        return self.container[identifier]

    def copy(self, *args, **kwargs) -> DependencyProvider:
        dp = LagomDependencyProvider(self.container.clone())
        dp.update(*args, **kwargs)
        return dp


dp1 = LagomDependencyProvider(application_container)

# make a copy
dp2 = dp1.copy()

# make sure that the original and the copy are the same
assert dp1[Name] == dp2[Name] == "Foo"
assert dp1[Engine] is dp2[Engine]

# create a copy with overriden value
dp3 = dp1.copy(name=as_type("Bar", Name))  # not yet implemented

# make sure that the original was not overriden
assert dp3[Name] == "Bar" and dp1[Name] == "Foo"


app = Application(dependency_provider=LagomDependencyProvider(application_container))


@app.on_create_transaction_context
def on_create_transaction_context():
    # if you want to share application container with transaction context use this:
    # return TransactionContext(app.dependency_provider)

    # if you want to have a separate container for transaction context use this:
    engine = app.dependency_provider[Engine]
    transaction_container = Container()
    transaction_container[CorrelationId] = uuid.uuid4()
    transaction_container[Session] = engine.create_sesson()
    transaction_container[Repository] = Repository(transaction_container[Session])
    dp = LagomDependencyProvider(transaction_container)
    return TransactionContext(dp)


@app.on_enter_transaction_context
def on_enter_transaction_context(ctx: TransactionContext):
    print("New transaction started", ctx)


def foo(correlation_id: CorrelationId, session: Session, repository: Repository):
    print(correlation_id, session, repository)


with app.transaction_context() as ctx:
    ctx.call(foo)
    ctx.call(foo)

with app.transaction_context() as ctx:
    ctx.call(foo)

dependency_injector integration

This code showcases a dependency provider based on dependency_injector:

import copy
import inspect
import uuid
from typing import Optional

from dependency_injector import containers, providers
from dependency_injector.containers import Container
from dependency_injector.providers import Dependency, Factory, Provider, Singleton
from dependency_injector.wiring import Provide, inject  # noqa

from lato import Application, DependencyProvider, TransactionContext


class Session:
    ...


class Repository:
    def __init__(self, session: Session):
        self.session = session


class Engine:
    def __init__(self, url):
        self.url = url

    def create_sesson(self):
        return Session()


class ApplicationContainer(containers.DeclarativeContainer):
    name = providers.Object("Foo")
    engine = providers.Singleton(Engine, "sqlite:///:memory:")


class TransactionContainer(containers.DeclarativeContainer):
    correlation_id = providers.Dependency(instance_of=uuid.UUID)
    session = providers.Dependency(instance_of=Session)
    repository = providers.Singleton(Repository, session=session)


def resolve_provider_by_type(container: Container, cls: type) -> Optional[Provider]:
    def inspect_provider(provider: Provider) -> bool:
        if isinstance(provider, (Factory, Singleton)):
            return issubclass(provider.cls, cls)
        elif isinstance(provider, Dependency):
            return issubclass(provider.instance_of, cls)

        return False

    matching_providers = inspect.getmembers(
        container,
        inspect_provider,
    )
    if matching_providers:
        if len(matching_providers) > 1:
            raise ValueError(
                f"Cannot uniquely resolve {cls}. Found {len(providers)} matching providers."
            )
        return matching_providers[0][1]
    return None


class ContainerProvider(DependencyProvider):
    def __init__(self, container: Container):
        self.container = container
        self.counter = 0

    def has_dependency(self, identifier: str | type) -> bool:
        if isinstance(identifier, type) and resolve_provider_by_type(
            self.container, identifier
        ):
            return True
        if type(identifier) is str:
            return identifier in self.container.providers

    def register_dependency(self, identifier, dependency_instance):
        pr = providers.Object(dependency_instance)
        try:
            setattr(self.container, identifier, pr)
        except TypeError:
            setattr(self.container, f"{str(identifier)}-{self.counter}", pr)
            self.counter += 1

    def get_dependency(self, identifier):
        try:
            if isinstance(identifier, type):
                provider = resolve_provider_by_type(self.container, identifier)
            else:
                provider = getattr(self.container, identifier)
            instance = provider()
        except Exception as e:
            raise e
        return instance

    def copy(self, *args, **kwargs):
        dp = ContainerProvider(copy.copy(self.container))
        dp.update(*args, **kwargs)
        return dp


# some tests...
ac = ApplicationContainer()

dp1 = ContainerProvider(ac)

# make a copy
dp2 = dp1.copy()

# make sure that the original and the copy are the same
assert dp1["name"] == dp2["name"] == "Foo"
assert dp1["engine"] is dp2["engine"]

# create a copy with overriden value
dp3 = dp1.copy(name="Bar")

# make sure that the original was not overriden
assert dp3["name"] == "Bar" and dp1["name"] == "Foo"


app = Application(dependency_provider=ContainerProvider(ApplicationContainer()))


@app.on_create_transaction_context
def on_create_transaction_context():
    # if you want to share application container with transaction context use this:
    # return TransactionContext(app.dependency_provider)

    # if you want to have a separate container for transaction context use this:
    engine = app.dependency_provider["engine"]
    dp = ContainerProvider(
        TransactionContainer(
            correlation_id=uuid.uuid4(), session=engine.create_sesson()
        )
    )
    return TransactionContext(dp)


@app.on_enter_transaction_context
def on_enter_transaction_context(ctx: TransactionContext):
    print("New transaction started")


def foo(correlation_id: uuid.UUID, session: Session, repository: Repository):
    print(correlation_id, session, repository)


with app.transaction_context() as ctx:
    ctx.call(foo)
    ctx.call(foo)

with app.transaction_context() as ctx:
    ctx.call(foo)