Extensions

Extensions are a mechanism to extend container and context behavior, similar to a plugin system.

Container Extensions

Lifespan

Lifespan extension could be used to execute code when container enters and exits

import asyncio
import contextlib
from collections.abc import AsyncIterator

from aioinject import Container
from aioinject.extensions import LifespanExtension


class MyLifespanExtension(LifespanExtension):
    @contextlib.asynccontextmanager
    async def lifespan(
        self,
        container: Container,  # noqa: ARG002
    ) -> AsyncIterator[None]:
        print("Enter")
        yield None
        print("Exit")


async def main() -> None:
    container = Container(extensions=[MyLifespanExtension()])
    async with container:
        # print("Enter") is executed.
        pass
        # print("Exit") is executed.


if __name__ == "__main__":
    asyncio.run(main())

OnInit

OnInit extension is executed when container's __init__ is called, this could be used for example to register dependencies in it:

from datetime import datetime
from typing import NewType

from aioinject import Container, Transient
from aioinject.extensions import OnInitExtension


Now = NewType("Now", datetime)


class TimeExtension(OnInitExtension):
    def on_init(
        self,
        container: Container,
    ) -> None:
        container.register(Transient(datetime.now, Now))


container = Container(extensions=[TimeExtension()])
with container.sync_context() as ctx:
    print(ctx.resolve(Now))

Context Extensions

Context extensions can be added to individual contexts when creating them

async with container.context(
    extensions=[MyContextExtension()],
):
    pass

OnResolve / OnResolveSync

On resolve extension is called when individual dependency is provided within a context:

import logging
from typing import TypeVar

from aioinject import InjectionContext, Provider
from aioinject.extensions import OnResolveExtension


T = TypeVar("T")


class MyExtension(OnResolveExtension):
    async def on_resolve(
        self,
        context: InjectionContext,  # noqa: ARG002
        provider: Provider[T],
        instance: T,  # noqa: ARG002
    ) -> None:
        logging.info("%s type was provided!", provider.type_)