Skip to main content

A dependency injection library for Python, Optimized for serverless applications

Project description

Ididi

Genius simplicity, unmatched power

ididi is 100% test covered and strictly typed.

codecov PyPI version Python Version License Downloads


Documentation: https://raceychan.github.io/ididi

Source Code: https://github.com/raceychan/ididi


Install

ididi requires python >= 3.9

pip install ididi

To view viusal dependency graph, install graphviz

pip install ididi[graphviz]

Features

  • No / minial changes to your existing code
  • Smart injection based on type hints with strong support to typing module.
  • Advanced scope support
  • Formated and detail-rich error messages
  • Highly performant, on part with nodejs and go

Usage

Quick Start

from typing import AsyncGenerator
from ididi import use, entry

async def conn_factory(engine: AsyncEngine) -> AsyncGenerator[AsyncConnection, None]:
    async with engine.begin() as conn:
        yield conn

class UnitOfWork:
    def __init__(self, conn: AsyncConnection=use(conn_factory)):
        self._conn = conn

@entry
async def main(command: CreateUser, uow: UnitOfWork):
    await uow.execute(build_query(command))

# note uow is automatically injected here
await main(CreateUser(name='user'))

Dependency factory

assign which factory method to use with ididi.use

from ididi import use
from datetime import datetime, timezone

def utc_factory() -> datetime:
    return datetime.now(timezone.utc)

UTC_DATETIME = Annotated[datetime, use(utc_factory)]

class Timer:
    def __init__(self, time: UTC_DATETIME):
        self.time = time

tmer = ididi.resolve(Timer)
assert tmer.time.tzinfo == timezone.utc

[!TIP] DependencyGraph.node accepts a wide arrange of types, such as dependent class, sync/async facotry, sync/async resource factory, with typing support.

Scope

Using Scope to manage resources

  • Infinite number of nested scope
  • Parent scope can be accssed by its child scopes(within the same context)
  • Resources will be shared across dependents only withint the same scope
  • Resources will be automatically closed and destroyed when the scope is exited.
  • Classes that implment contextlib.AbstractContextManager or contextlib.AbstractAsyncContextManager are also considered to be resources and can/should be resolved within scope.
  • Scopes are separated by context

[!TIP] If you have two call stack of a1 -> b1 and a2 -> b2, Here a1 and a2 are two calls to the same function a, then, in b1, you can only access scope created by the a1, not a2.

This is particularly useful when you try to separate resources by route, endpoint, request, etc.

Async scope

@dg.node
def get_resource() -> ty.Generator[Resource, None, None]:
    res = Resource()
    with res:
        yield res

@dg.node
async def get_asyncresource() -> ty.Generator[AsyncResource, None, None]:
    res = AsyncResource()
    async with res:
        yield res


with dg.scope() as scope:
    resource = scope.resolve(Resource)

# For async generator
async with dg.scope() as scope:
    resource = await scope.resolve(AsyncResource)

[!TIP] dg.node will leave your class/factory untouched, i.e., you can use it just like it is not decorated.

Contexted Scope

You can use dg.use_scope to retrive most recent scope, context-wise, this allows your to have access the scope without passing it around, e.g.

async def service_factory():
    async with dg.scope() as scope:
        service = scope.resolve(Service)
        yield service

@app.get("users")
async def get_user(service: Service = Depends(service_factory))
    await service.create_user(...)

Then somewhere deep in your service.create_user call stack

async def create_and_publish():
    uow = dg.use_scope().resolve(UnitOfWork)
    async with uow.trans():
        user_repo.add_user(user)
        event_store.add(user_created_event)

Here dg.use_scope() would return the same scope you created in your service_factory.

Named Scope

You can create infinite level of scopes by assigning hashable name to scopes

# at the top most entry of a request
async with dg.scope(request_id) as scope:
    ...

now scope with name request_id is accessible everywhere within the request context

request_scope = dg.use_scope(request_id)

[!NOTE] Two or more scopes with the same name would follow most recent rule.

Nested Nmaed Scope

async with dg.scope(app_name) as app_scope:
    async with dg.scope(router_name) as router_scope:
        async with dg.scope(endpoint_name) as endpoint_scope:
            async with dg.scope(user_id) as user_scope:
                async with dg.scope(request_id) as request_scope:
                    ...

For any functions called within the request_scope, you can get the most recent scope with dg.use_scope(), or its parent scopes, i.e. dg.use_scope(app_name) to get app_scope.

Testing

Override class dependency resolution

You can control how ididi resolve a dependency during testing, by register the test double of the dependency using dg.node

Example: For the following dependent

class UserRepository:
    def __init__(self, db: DataBase):
        self.db=db

dg = DependencyGraph()
assert isinstance(dg.resolve(UserRepository).db, DataBase)

in you test file,

class FakeDB(DataBase): ...

def db_factory() -> DataBase:
    return FakeDB()


def test_resolve():
    dg = DependencyGraph()
    dg.node(reuse=False)(UserRepository)
    assert isinstance(dg.resolve(UserRepository).db, DataBase)

    dg.node(db_factory)
    assert isinstance(dg.resolve(UserRepository).db, FakeDB)

use dg.node to replace DataBase with its test double.

Override entry dependency resolution

async def test_entry_replace():
    dg = DependencyGraph()

    async def create_user(
        user_name: str, user_email: str, service: UserService
    ) -> UserService:
        return service

    class FakeUserService(UserService): ...

    create_user = dg.entry(reuse=False)(create_user)
    create_user.replace(UserService, FakeUserService)

    res = await create_user("user", "user@email.com")
    assert isinstance(res, FakeUserService)

use entryfunc.replace to replace a dependency with its test double. you can also use replace(service=FakeUserService) to override

More

For more detailed information, check out Documentation

  • Tutorial

  • Usage of factory

  • Visualize the dependency graph(beta)

  • Circular Dependency Detection

  • Error context

Project details


Release history Release notifications | RSS feed

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

ididi-1.3.0.tar.gz (98.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

ididi-1.3.0-py3-none-any.whl (28.6 kB view details)

Uploaded Python 3

File details

Details for the file ididi-1.3.0.tar.gz.

File metadata

  • Download URL: ididi-1.3.0.tar.gz
  • Upload date:
  • Size: 98.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.0

File hashes

Hashes for ididi-1.3.0.tar.gz
Algorithm Hash digest
SHA256 8f3bd94af9fb6c60338e9d8642c0e3c99d22c261fe2424a60ad918ae2455f28b
MD5 c8e73290f350ea1d790bcdcb1ed69a55
BLAKE2b-256 05a2dd52888e12cfdca600618b8c466ecc52bbf8071b86d68ec1e65654f256b5

See more details on using hashes here.

File details

Details for the file ididi-1.3.0-py3-none-any.whl.

File metadata

  • Download URL: ididi-1.3.0-py3-none-any.whl
  • Upload date:
  • Size: 28.6 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/5.1.1 CPython/3.9.0

File hashes

Hashes for ididi-1.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 34ce982035a8879f3e6da0dde86e4e8094ca295ab4fc3e491933b0ad62a4140c
MD5 d7f38ecbfcb5dda4100253e12817e268
BLAKE2b-256 c80adc18ac95582bd2f5e54c0653c0665df25033d4eb8e2714c36cb15b9ff0c2

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page