Injection

Easy dependency injection for all, works with Python 3.8-3.13. Main features and advantages:

• support Python 3.8-3.13;
• works with FastAPI, Litestar, Flask and Django REST Framework;
• support dependency injection via Annotated in FastAPI;
• support async injections;
• support auto injection by types;
• resources with function scope;
• no wiring;
• overriding dependencies for testing;
• 100% code coverage;
• the code is fully typed and checked with mypy;
• good documentation;
• intuitive and almost identical api with dependency-injector, which will allow you to easily migrate to injection (see migration from dependency injector);

Installation

pip install deps-injection

Compatibility between web frameworks and injection features

Quickstart with FastAPI, SQLAlchemy and pytest (sync sqlite)

from contextlib import contextmanager
from random import Random
from typing import Annotated, Any, Callable, Dict, Iterator

import pytest
from fastapi import Depends, FastAPI
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker
from starlette.testclient import TestClient

from injection import DeclarativeContainer, Provide, inject, providers


@contextmanager
def db_session_resource(session_factory: Callable[..., Session]) -> Iterator[Session]:
    session = session_factory()
    try:
        yield session
    except Exception:
        session.rollback()
    finally:
        session.close()


class SomeDAO:
    def __init__(self, db_session: Session) -> None:
        self.db_session = db_session

    def get_some_data(self, num: int) -> int:
        stmt = text("SELECT :num").bindparams(num=num)
        data: int = self.db_session.execute(stmt).scalar_one()
        return data


class DIContainer(DeclarativeContainer):
    db_engine = providers.Singleton(
        create_engine,
        url="sqlite:///db.db",
        pool_size=20,
        max_overflow=0,
        pool_pre_ping=False,
    )

    session_factory = providers.Singleton(
        sessionmaker,
        db_engine.cast,
        autoflush=False,
        autocommit=False,
    )

    db_session = providers.Resource(
        db_session_resource,
        session_factory=session_factory.cast,
        function_scope=True,
    )

    some_dao = providers.Factory(SomeDAO, db_session=db_session.cast)


SomeDAODependency = Annotated[SomeDAO, Depends(Provide[DIContainer.some_dao])]

app = FastAPI()


@app.get("/values/{value}")
@inject
async def sqla_resource_handler_async(
    value: int,
    some_dao: SomeDAODependency,
) -> Dict[str, Any]:
    value = some_dao.get_some_data(num=value)
    return {"detail": value}


@pytest.fixture(scope="session")
def test_client() -> TestClient:
    client = TestClient(app)
    return client


def test_sqla_resource(test_client: TestClient) -> None:
    rnd = Random()
    random_int = rnd.randint(-(10**6), 10**6)

    response = test_client.get(f"/values/{random_int}")

    assert response.status_code == 200
    assert not DIContainer.db_session.initialized
    body = response.json()
    assert body["detail"] == random_int

Quickstart with FastAPI, SQLAlchemy and pytest (async sqlite)

from contextlib import asynccontextmanager
from random import Random
from typing import Annotated, Any, Callable, Dict, AsyncIterator

import pytest
from fastapi import Depends, FastAPI
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from starlette.testclient import TestClient

from injection import DeclarativeContainer, Provide, inject, providers


@asynccontextmanager
async def db_session_resource(session_factory: Callable[..., AsyncSession]) -> AsyncIterator[AsyncSession]:
    session = session_factory()
    try:
        yield session
    except Exception:
        await session.rollback()
    finally:
        await session.close()


class SomeDAO:
    def __init__(self, db_session: AsyncSession) -> None:
        self.db_session = db_session

    async def get_some_data(self, num: int) -> int:
        stmt = text("SELECT :num").bindparams(num=num)
        result = await self.db_session.execute(stmt)
        data: int = result.scalar_one()
        return data


class DIContainer(DeclarativeContainer):
    # need to install aiosqlite and greenlet
    db_engine = providers.Singleton(
        create_async_engine,
        url="sqlite+aiosqlite:///db.db",
        pool_pre_ping=False,
    )

    session_factory = providers.Singleton(
        async_sessionmaker,
        db_engine.cast,
        autoflush=False,
        autocommit=False,
    )

    db_session = providers.Resource(
        db_session_resource,
        session_factory=session_factory.cast,
        function_scope=True,
    )

    some_dao = providers.Factory(SomeDAO, db_session=db_session.cast)


SomeDAODependency = Annotated[SomeDAO, Depends(Provide[DIContainer.some_dao])]

app = FastAPI()


@app.get("/values/{value}")
@inject
async def sqla_resource_handler_async(
        value: int,
        some_dao: SomeDAODependency,
) -> Dict[str, Any]:
    value = await some_dao.get_some_data(num=value)
    return {"detail": value}


@pytest.fixture(scope="session")
def test_client() -> TestClient:
    client = TestClient(app)
    return client


def test_async_sqla_resource(test_client: TestClient) -> None:
    rnd = Random()
    random_int = rnd.randint(-(10 ** 6), 10 ** 6)

    response = test_client.get(f"/values/{random_int}")

    assert response.status_code == 200
    assert not DIContainer.db_session.initialized
    body = response.json()
    assert body["detail"] == random_int