Async-first dependency injection library based on python type hints

Overview

Dependency Depression

Async-first dependency injection library based on python type hints

Quickstart

First let's create a class we would be injecting:

class Test:
    pass

Then we should create instance of container and register our Test class in it, we would use Callable provider that would simply call our class, since classes are also callables!

from dependency_depression import Depression, Callable

container = Depression()
container.register(Test, Callable(Test))

Then we should create a context and resolve our class from it:

with container.sync_context() as ctx:
    ctx.resolve(Test)
    # < __main__.Test>

Injecting

To mark parameters for injection mark them with typing.Annotated and Inject marker

from typing import Annotated
from dependency_depression import Callable, Depression, Inject


def create_number() -> int:
    return 42


def create_str(number: Annotated[int, Inject]) -> str:
    return str(number)

container = Depression()
container.register(str, Callable(create_str))
container.register(int, Callable(create_number))

with container.sync_context() as ctx:
    string = ctx.resolve(str)
    print(string, type(string))
    # 42 
   

Providers

When creating a provider you should specify the type it returns, but it can be inferred from class type or function return type:

from dependency_depression import Callable

provider = Callable(int)
# Is the same as Callable(factory=int, impl=int)
assert provider.provide_sync() == 0

Example using factory function, impl is inferred from return type hint:

from dependency_depression import Callable


def create_foo() -> str:
    return "foo"


provider = Callable(create_foo)
assert provider.provide_sync() == "foo"
assert provider.impl is str

This all comes into play when you have multiple implementations for base class and want to retrieve individual providers from a container,
let's register two concrete classes under same interface:

from dependency_depression import Depression, Callable


class Base:
    pass


class ConcreteA(Base):
    pass


class ConcreteB(Base):
    pass


container = Depression()
container.register(Base, Callable(ConcreteA))
container.register(Base, Callable(ConcreteB))

with container.sync_context() as ctx:
    a = ctx.resolve(Base, ConcreteA)  # <__main__.ConcreteA>
    b = ctx.resolve(Base, ConcreteB)  # <__main__.ConcreteB>
    
    # This would raise an error since we have two classes registered as `Base`
    ctx.resolve(Base)

If you have multiple classes registered under same interface you can specify concrete class using Impl marker:

from typing import Annotated
from dependency_depression import Inject, Impl


class Injectee:
    def __init__(
        self,
        a: Annotated[Base, Inject, Impl[ConcreteA]],
        b: Annotated[Base, Inject, Impl[ConcreteB]],
    ):
        pass

You can also just register concrete classes instead:

container.register(ConcreteA, Callable(ConcreteA))
container.register(ConcreteB, Callable(ConcreteB))

class Injectee:
    def __init__(
        self,
        a: Annotated[ConcreteA, Inject],
        b: Annotated[ConcreteB, Inject],
    ):
        pass

Generics

Dependency Depression can also be used with Generics:

T: raise NotImplementedError class UserRepository(IRepository[User]): def get(self, identity: int) -> User: return User(id=identity, username="Username") class ItemRepository(IRepository[Item]): def get(self, identity: int) -> Item: return Item(id=identity, title="Title") class Injectee: def __init__( self, user_repository: Annotated[IRepository[User], Inject], item_repository: Annotated[IRepository[Item], Inject], ): self.user_repository = user_repository self.item_repository = item_repository container = Depression() container.register(IRepository[User], Callable(UserRepository)) container.register(IRepository[Item], Callable(ItemRepository)) container.register(Injectee, Callable(Injectee)) with container.sync_context() as ctx: injectee = ctx.resolve(Injectee) injectee.user_repository # < __main__.UserRepository> injectee.item_repository # <__main__.ItemRepository>">
import dataclasses
from typing import Generic, TypeVar, Annotated

from dependency_depression import Inject, Depression, Callable

T = TypeVar("T")


@dataclasses.dataclass
class User:
    id: int
    username: str


@dataclasses.dataclass
class Item:
    id: int
    title: str


class IRepository(Generic[T]):
    def get(self, identity: int) -> T:
        raise NotImplementedError


class UserRepository(IRepository[User]):
    def get(self, identity: int) -> User:
        return User(id=identity, username="Username")

    
class ItemRepository(IRepository[Item]):
    def get(self, identity: int) -> Item:
        return Item(id=identity, title="Title")

    
class Injectee:
    def __init__(
        self,
        user_repository: Annotated[IRepository[User], Inject],
        item_repository: Annotated[IRepository[Item], Inject],
    ):
        self.user_repository = user_repository
        self.item_repository = item_repository


container = Depression()
container.register(IRepository[User], Callable(UserRepository))
container.register(IRepository[Item], Callable(ItemRepository))
container.register(Injectee, Callable(Injectee))

with container.sync_context() as ctx:
    injectee = ctx.resolve(Injectee)
    injectee.user_repository
    # < __main__.UserRepository>
    injectee.item_repository
    # <__main__.ItemRepository>

Context

Context as meant to be used within application or request scope, it keeps instances cache and an ExitStack to close all resources.

Cache

Context keeps cache of all instances, so they won't be created again, unless use_cache=False or NoCache is used.

In this example passing use_cache=False would cause context to create instance of Test again, however it wouldn't be cached:

from dependency_depression import Callable, Depression


class Test:
    pass


container = Depression()
container.register(Test, Callable(Test))

with container.sync_context() as ctx:
    first = ctx.resolve(Test)
    
    assert first is not ctx.resolve(Test, use_cache=False)
    # first is still cached in context
    assert first is ctx.resolve(Test)

Closing resources using context managers

Context would also use functions decorated with contextlib.contextmanager or contextlib.asyncontextmanager, but it won't use other instances of ContextManager.
Note that you're not passing impl parameter should specify return type using Iterable, Generator or their async counterparts - AsyncIterableand AsyncGenerator:

import contextlib
from typing import Iterable

from dependency_depression import Depression, Callable


@contextlib.contextmanager
def contextmanager() -> Iterable[int]:
    yield 42


class ContextManager:
    def __enter__(self):
        # This would never be called
        raise ValueError

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


container = Depression()

# Without return type hint you can specify impl parameter:
# container.register(int, Callable(contextmanager, int))
container.register(int, Callable(contextmanager))
container.register(ContextManager, Callable(ContextManager))

with container.sync_context() as ctx:
    number = ctx.resolve(int)  # 42
    ctx_manager = ctx.resolve(ContextManager) # __enter__ would not be called
    with ctx_manager:
        ...
        # Oops, ValueError raised

In case you need to manage lifecycle of your objects you should wrap them in a context manager:

import contextlib
from typing import AsyncGenerator

from dependency_depression import Callable, Depression
from sqlalchemy.ext.asyncio import AsyncSession


@contextlib.asynccontextmanager
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    session = AsyncSession()
    async with session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

container = Depression()
container.register(AsyncSession, Callable(AsyncSession))

@Inject decorator

@inject decorator allows you to automatically inject parameters into functions:

from typing import Annotated

from dependency_depression import Callable, Depression, Inject, inject


@inject
def injectee(number: Annotated[int, Inject]):
    return number


container = Depression()
container.register(int, Callable(int))

with container.sync_context():
    print(injectee())
    # 0

Without active context number parameter would not be injected:

injectee()
# TypeError: injectee() missing 1 required positional argument: 'number'

But you still can use your function just fine

print(injectee(42))

You can pass parameters even if you have an active context:

with container.sync_context():
    print(injectee())  # 0, injected
    print(injectee(42))  # 42, provided by user

Usage with Asyncio

Dependency Depression can be used in async context, just use context instead of sync_context:

import asyncio

from dependency_depression import Callable, Depression


async def get_number() -> int:
    await asyncio.sleep(1)
    return 42


async def main():
    container = Depression()
    container.register(int, Callable(get_number))
    async with container.context() as ctx:
        number = await ctx.resolve(int)
        assert number == 42


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

Async context also supports both sync and async context managers and factory functions.

You might also like...
nbsafety adds a layer of protection to computational notebooks by solving the stale dependency problem when executing cells out-of-order
nbsafety adds a layer of protection to computational notebooks by solving the stale dependency problem when executing cells out-of-order

nbsafety adds a layer of protection to computational notebooks by solving the stale dependency problem when executing cells out-of-order

Pipenv-local-deps-repro - Reproduction of a local transitive dependency on pipenv

Reproduction of the pipenv bug with transitive local dependencies. Clone this re

An async API wrapper for Dress To Impress written in Python.

dti.py An async API wrapper for Dress To Impress written in Python. Some notes: For the time being, there are no front-facing docs for this beyond doc

Async Python Circuit Breaker implementation

aiocircuitbreaker This is an async Python implementation of the circuitbreaker library. Installation The project is available on PyPI. Simply run: $ p

This is a survey of python's async concurrency features by example.

Survey of Python's Async Features This is a survey of python's async concurrency features by example. The purpose of this survey is to demonstrate tha

The first Python 1v1.lol triggerbot working with colors !
The first Python 1v1.lol triggerbot working with colors !

1v1.lol TriggerBot Afin d'utiliser mon triggerbot, vous devez activer le plein écran sur 1v1.lol sur votre naviguateur (quelque-soit ce dernier). Vous

Recreating my first CRUD in python, but now more professional

Recreating my first CRUD in python, but now more professional

Type Persian without confusing words for yourself and others, in Adobe Connect
Type Persian without confusing words for yourself and others, in Adobe Connect

About In the Adobe Connect chat section, to type in Persian or Arabic, the written words will be confused and will be written and sent illegibly (This

Handwrite - Type in your Handwriting!
Handwrite - Type in your Handwriting!

Handwrite - Type in your Handwriting! Ever had those long-winded assignments, that the teacher always wants handwritten?

Owner
Doctor
Doctor
Force you (or your user) annotate Python function type hints.

Must-typing Force you (or your user) annotate function type hints. Notice: It's more like a joke, use it carefully. If you call must_typing in your mo

Konge 13 Feb 19, 2022
Pre-commit hook for upgrading type hints

This is a pre-commit hook configured to automatically upgrade your type hints to the new native types implemented in PEP 585.

snok 54 Nov 14, 2022
Simple dependency injection framework for Python

A simple, strictly typed dependency injection library.

BentoML 14 Jun 29, 2022
Async timeit - Async version of python's timeit

Async Timeit Replica of default python timeit module with small changes to allow

Raghava G Dhanya 3 Apr 13, 2022
edgetest is a tox-inspired python library that will loop through your project's dependencies, and check if your project is compatible with the latest version of each dependency

Bleeding edge dependency testing Full Documentation edgetest is a tox-inspired python library that will loop through your project's dependencies, and

Capital One 16 Dec 7, 2022
Runtime fault injection platform by Daniele Rizzieri (2021)

GDBitflip [v1.04] Runtime fault injection platform by Daniele Rizzieri (2021) This platform executes N times a binary and during each execution it inj

Daniele Rizzieri 1 Dec 7, 2021
Runtime fault injection platform by Daniele Rizzieri

GDBitflip [v1.04] Runtime fault injection platform by Daniele Rizzieri (2021) This platform executes N times a binary and during each execution it inj

Daniele Rizzieri 1 Dec 7, 2021
🗽 Like yarn outdated/upgrade, but for pip. Upgrade all your pip packages and automate your Python Dependency Management.

pipupgrade The missing command for pip Table of Contents Features Quick Start Usage Basic Usage Docker Environment Variables FAQ License Features Upda

Achilles Rasquinha 529 Dec 31, 2022
A python script developed to process Windows memory images based on triage type.

Overview A python script developed to process Windows memory images based on triage type. Requirements Python3 Bulk Extractor Volatility2 with Communi

CrowdStrike 245 Nov 24, 2022
This is a a CSMA/CA simulator written in Python based on simulator of the same type

This is a a CSMA/CA simulator written in Python based on simulator of the same type found the link https://github.com/StevenSLXie/CSMA-Simulator with

M. Ismail 4 Nov 22, 2022