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.

Owner
Doctor
Doctor
Um pequeno painel de consulta grátis.

[PAINEL-DE-CONSULTA 3.8(BETA)] · Confira meu canal do YouTube. Clique aqui! Nota: Próxima Atualização será a última com coisas novas, o resto será par

276 Jan 05, 2023
Like Docker, but for Squeak. You know, for kids.

Squeaker Like Docker, but for Smalltalk images. You know, for kids. It's a small program that helps in automated derivation of configured Smalltalk im

Tony Garnock-Jones 14 Sep 11, 2022
A comparison of mesh generators.

This repository creates meshes of the same domains with multiple mesh generators and compares the results.

Nico Schlömer 29 Dec 12, 2022
Show my read on kindle this year

Show my kindle status on GitHub

yihong 26 Jun 20, 2022
Module 2's katas from Launch X's python introduction course.

Module2Katas Module 2's katas from Launch X's python introduction course. Virtual environment creation process (on Windows): Create a folder in any de

Javier Méndez 1 Feb 10, 2022
Controller state monitor plugin for EVA ICS

eva-plugin-cmon Controller status monitor plugin for EVA ICS Monitors connected controllers status in SFA and pushes measurements into an external Inf

Altertech 1 Nov 06, 2021
Automatización del proceso Inmofianza

Selenium Inmofianza Proyecto de pruebas automatizadas con selenium webdriver para el aplicativo Omnicanalidad Pre-requisitos 📋 Componentes que deben

Natalia Narváez 1 Jan 07, 2022
This is a Blender 2.9 script for importing mixamo Models to Godot-3

Mixamo-To-Godot This is a Blender 2.9 script for importing mixamo Models to Godot-3 The script does the following things Imports the mixamo models fro

8 Sep 02, 2022
A utility control surface for Ableton Live that makes the initialization of a Mixdown quick

Automate Mixdown initialization A script that transfers all the VSTs on your MIDI tracks to a new track so you can freeze your MIDI tracks and then co

Aarnav 0 Feb 23, 2022
Customizable-menu-python - User customizable menu in Python

Menu personalizável pelo usuário em Python A minha ideia com esse projeto pessoa

Renan Barbosa 4 Oct 28, 2022
Whatsapp Messenger master

Whatsapp Messenger master

Swarup Kharul 5 Nov 21, 2021
Project aims to map out common user behavior on the computer

User-Behavior-Mapping-Tool Project aims to map out common user behavior on the computer. Most of the code is based on the research by kacos2000 found

trustedsec 136 Dec 23, 2022
Mmr image postbot - Бот для создания изображений с новыми релизами в сообщество ВК MMR Aggregator

Mmr image postbot - Бот для создания изображений с новыми релизами в сообщество ВК MMR Aggregator

Max 3 Jan 07, 2022
This alerts you when the avalanche score a goal

This alerts you when the avalanche score a goal

Davis Burrill 1 Jan 15, 2022
This is a Docker-based pipeline for preparing sextractor-ready multiwavelength images

Pipeline for creating NB422-detected (ODI) catalog The repository contains a Docker-based pipeline for preprocessing observational data. The pipeline

1 Sep 01, 2022
Konomi: Kind and Optimized Next brOadcast watching systeM Infrastructure

Konomi 備考・注意事項 現在 α 版で、まだ実験的なプロダクトです。通常利用には耐えないでしょうし、サポートもできません。 安定しているとは到底言いがたい品質ですが、それでも構わない方のみ導入してください。 使い方などの説明も用意できていないため、自力でトラブルに対処できるエンジニアの方以外に

tsukumi 243 Dec 30, 2022
DOP-Tuning(Domain-Oriented Prefix-tuning model)

DOP-Tuning DOP-Tuning(Domain-Oriented Prefix-tuning model)代码基于Prefix-Tuning改进. Files ├── seq2seq # Code for encoder-decoder arch

Andrew Zeng 5 Nov 02, 2022
A python script for combining multiple native SU2 format meshes into one mesh file for multi-zone simulations.

A python script for combining multiple native SU2 format meshes into one mesh file for multi-zone simulations.

MKursatUzuner 1 Jan 20, 2022
Usos Semester average helper

Usos Semester average helper Dzieki temu skryptowi mozesz sprawdzic srednia ocen na kazdy odbyty przez ciebie semestr PARAMETERS required: '--username

2 Jan 17, 2022
Fully coded Apps by Codex.

OpenAI-Codex-Code-Generation Fully coded Apps by Codex. How I use Codex in VSCode to generate multiple completions with autosorting by highest "mean p

nanowell 47 Jan 01, 2023