aiodine

Note: you are currently viewing the development branch for aiodine 2.0. For the documentation from the latest 1.x release, please see the latest branch.

aiodine provides a simple but powerful dependency injection mechanism for Python 3.6+ asynchronous programs.

Features

Simple and elegant API.
Setup/teardown logic via async context managers.
Dependency caching (coming soon).
Great typing support.
Compatible with asyncio, trio and curio.
100% test coverage.
Fully type-annotated.

Contents

Quickstart
Installation
User guide
FAQ
Changelog

Quickstart

import aiodine

async def moo() -> str:
    print("What does the cow say?")
    return "moo!"

async def cowsay(what: str = aiodine.depends(moo)):
    print(f"Going to say {what!r}...")
    print(f"Cow says {what}")

import trio
trio.run(aiodine.call_resolved, cowsay)

Output:

What does the cow say?
Going to say 'moo!'...
Cow says moo!

Running with asyncio or curio instead:

import asyncio
# Python 3.7+
asyncio.run(aiodine.call_resolved(main))
# Python 3.6
loop = asyncio.get_event_loop()
loop.run_until_complete(aiodine.call_resolved(main))

import curio
curio.run(aiodine.call_resolved, (main,))

Installation

pip install aiodine

User guide

This section will be using trio as a concurrency library. Feel free to adapt the code for asyncio or curio.

Let's start with some imports...

import typing
import trio
import aiodine

Core ideas

The core concept in aiodine is that of a dependable.

A dependable is created by calling aiodine.depends(...):

async def cowsay(what: str) -> str:
    return f"Cow says {what}"

dependable = aiodine.depends(cowsay)

Let's inspect what the dependable refers to:

print(dependable)  # Dependable(func=<function cowsay at ...>)

Yup, looks good.

A dependable can't do much on its own — we need to use it along with call_resolved(), the main entry point in aiodine.

By default, call_resolved() acts as a proxy, i.e. it passes any positional and keyword arguments along to the given function:

async def main() -> str:
    return await aiodine.call_resolved(cowsay, what="moo")

assert trio.run(main) == "Cow says moo"

But call_resolved() can also inject dependencies into the function it is given. Put differently, call_resolved() does all the heavy lifting to provide the function with the arguments it needs.

async def moo() -> str:
    print("Evaluating 'moo()'...")
    await trio.sleep(0.1)  # Simulate some I/O...
    print("Done!")
    return "moo"

async def cowsay(what: str = aiodine.depends(moo)) -> str:
    print(f"cowsay got what={what!r}")
    return f"Cow says {what}"

async def main() -> str:
    # Note that we're leaving out the 'what' argument here.
    return await aiodine.call_resolved(cowsay)

print(trio.run(main)

This code will output the following:

Evaluating 'moo()'...
Done!
cowsay got what='moo'
Cow says moo

We can still pass arguments from the outside, in which case aiodine won't need to resolve anything.

For example, replace the content of main() with:

await aiodine.call_resolved(cowsay, "MOO!!")

It should output the following:

cowsay got what='MOO!!'
Cow says MOO!!

Typing support

You may have noticed that we used type annotations in the code snippets above. If you run the snippets through a static type checker such as mypy, you shouldn't get any errors.

On the other hand, if you change the type hint of what to, for example, int, then mypy will complain because types don't match anymore:

async def cowsay(what: int = aiodine.depends(moo)) -> str:
    return f"Cow says {what}"

Incompatible default for argument "what" (default has type "str", argument has type "int")

All of this is by design: aiodine tries to be as type checker-friendly as it can. It even has a test for the above situation!

Usage with context managers

Sometimes, the dependable has some setup and/or teardown logic associated with it. This is typically the case for most I/O resources such as sockets, files, or database connections.

This is why aiodine.depends() also accepts asynchronous context managers:

import typing
import aiodine

# On 3.7+, use `from contextlib import asynccontextmanager`.
from aiodine.compat import asynccontextmanager


class Database:
    def __init__(self, url: str) -> None:
    self.url = url

    async def connect(self) -> None:
        print(f"Connecting to {self.url!r}")

    async def fetchall(self) -> typing.List[dict]:
        print("Fetching data...")
        return [{"id": 1}]

    async def disconnect(self) -> None:
        print(f"Releasing connection to {self.url!r}")


@asynccontextmanager
async def get_db() -> typing.AsyncIterator[Database]:
    db = Database(url="sqlite://:memory:")
    await db.connect()
    try:
        yield db
    finally:
        await db.disconnect()


async def main(db: Database = aiodine.depends(get_db)) -> None:
    rows = await db.fetchall()
    print("Rows:", rows)


trio.run(aiodine.call_resolved, main)

This code will output the following:

Connecting to 'sqlite://:memory:'
Fetching data...
Rows: [{'id': 1}]
Releasing connection to 'sqlite://:memory:'

FAQ

Why "aiodine"?

aiodine contains "aio" as in asyncio, and "di" as in Dependency Injection. The last two letters end up making aiodine pronounce like iodine, the chemical element.

Changelog

See CHANGELOG.md.

License

MIT

dizballanze/aiodine