Run async workflows using pytest-fixtures-style dependency injection

Simon Willison

Last update: Jun 26, 2022

Related tags

General Utilities asyncio

Overview

asyncinject

Run async workflows using pytest-fixtures-style dependency injection

Installation

Install this library using pip:

$ pip install asyncinject

Usage

This library is inspired by pytest fixtures.

The idea is to simplify executing parallel asyncio operations by allowing them to be collected in a class, with the names of parameters to the class methods specifying which other methods should be executed first.

This then allows the library to create and execute a plan for executing various dependent methods in parallel.

Here's an example, using the httpx HTTP library.

from asyncinject import AsyncInjectAll
import httpx

async def get(url):
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text

class FetchThings(AsyncInjectAll):
    async def example(self):
        return await get("http://www.example.com/")

    async def simonwillison(self):
        return await get("https://simonwillison.net/search/?tag=empty")

    async def both(self, example, simonwillison):
        return example + "\n\n" + simonwillison


combined = await FetchThings().both()
print(combined)

If you run this in ipython (which supports top-level await) you will see output that combines HTML from both of those pages.

The HTTP requests to www.example.com and simonwillison.net will be performed in parallel.

The library will notice that both() takes two arguments which are the names of other async def methods on that class, and will construct an execution plan that executes those two methods in parallel, then passes their results to the both() method.

Parameters are passed through

Your dependent methods can require keyword arguments which are passed to the original method.

class FetchWithParams(AsyncInjectAll):
    async def get_param_1(self, param1):
        return await get(param1)

    async def get_param_2(self, param2):
        return await get(param2)

    async def both(self, get_param_1, get_param_2):
        return get_param_1 + "\n\n" + get_param_2


combined = await FetchWithParams().both(
    param1 = "http://www.example.com/",
    param2 = "https://simonwillison.net/search/?tag=empty"
)
print(combined)

Parameters with default values are ignored

You can opt a parameter out of the dependency injection mechanism by assigning it a default value:

class IgnoreDefaultParameters(AsyncInjectAll):
    async def go(self, calc1, x=5):
        return calc1 + x

    async def calc1(self):
        return 5

print(await IgnoreDefaultParameters().go())
# Prints 10

AsyncInject and @inject

The above example illustrates the AsyncInjectAll class, which assumes that every async def method on the class should be treated as a dependency injection method.

You can also specify individual methods using the AsyncInject base class an the @inject decorator:

from asyncinject import AsyncInject, inject

class FetchThings(AsyncInject):
    @inject
    async def example(self):
        return await get("http://www.example.com/")

    @inject
    async def simonwillison(self):
        return await get("https://simonwillison.net/search/?tag=empty")

    @inject
    async def both(self, example, simonwillison):
        return example + "\n\n" + simonwillison

The resolve() function

If you want to execute a set of methods in parallel without defining a third method that lists them as parameters, you can do so using the resolve() function. This will execute the specified methods (in parallel, where possible) and return a dictionary of the results.

from asyncinject import resolve

fetcher = FetchThings()
results = await resolve(fetcher, ["example", "simonwillison"])

results will now be:

{
    "example": "contents of http://www.example.com/",
    "simonwillison": "contents of https://simonwillison.net/search/?tag=empty"
}

Development

To contribute to this library, first checkout the code. Then create a new virtual environment:

cd asyncinject
python -m venv venv
source venv/bin/activate

Or if you are using pipenv:

pipenv shell

Now install the dependencies and test dependencies:

pip install -e '.[test]'

To run the tests:

pytest

Comments

Concurrency is not being optimized

It looks like concurrency / parallelism is not being maximized due to the grouping of dependencies into node groups. Here's a simple example:

import asyncio
from time import time
from typing import Annotated

async def a():
    await asyncio.sleep(1)

async def b():
    await asyncio.sleep(2)

async def c(a):
    await asyncio.sleep(1)

async def d(b, c):
    pass

async def main_asyncinjector():
    reg = Registry(a, b, c, d)
    start = time()
    await reg.resolve(d)
    print(time()-start)

asyncio.run(main_asyncinjector())

This should take 2 seconds to run (start a and b, once a finishes start c, b and c finish at the same time and you're done) but takes 3 seconds (start a and b, wait for both to finish then start c).

This happens because graphlib.TopologicalSorter is not used online and instead it is being used to statically compute groups of dependencies.

I don't think it would be too hard to address this, but I'm not sure how much you'd want to change to accommodate this. I work on a similar project (https://github.com/adriangb/di) and there I found it very useful to break out the concept of an "executor" out of the container/registry concept, which means that instead of a parallel option you'd have pluggable executors that could choose to use concurrency, limit concurrency, use threads instead, etc. FWIW here's what that looks like with this example:

import asyncio
from time import time
from typing import Annotated

from asyncinject import Registry
from di.dependant import Marker, Dependant
from di.container import Container
from di.executors import ConcurrentAsyncExecutor


async def a():
    await asyncio.sleep(1)

async def b():
    await asyncio.sleep(2)

async def c(a: Annotated[None, Marker(a)]):
    await asyncio.sleep(1)

async def d(b: Annotated[None, Marker(b)], c: Annotated[None, Marker(c)]):
    pass

async def main_asyncinjector():
    reg = Registry(a, b, c, d)
    start = time()
    await reg.resolve(d)
    print(time()-start)


async def main_di():
    container = Container()
    solved = container.solve(Dependant(d), scopes=[None])
    executor = ConcurrentAsyncExecutor()
    async with container.enter_scope(None) as state:
        start = time()
        await container.execute_async(solved, executor, state=state)
        print(time()-start)

asyncio.run(main_asyncinjector())  # 3 seconds
asyncio.run(main_di())  # 2 seconds

enhancement

opened by adriangb 5

Investigate a non-class-based version

I'm thinking about using this with Datasette plugins, which aren't well suited to the current class-based mechanism because plugins may want to register their own additional dependency injection functions.
research

opened by simonw 4
Debug mechanism

Add a mechanism which shows exactly how the class is executing, including which methods are running in parallel. Maybe even with a very basic ASCII visualization? Then use it to help illustrate the examples in the README, refs #4.
enhancement

opened by simonw 4
A way to turn off parallel execution (for easier comparison)

Would be neat if you could toggle the parallel execution on and off, to better demonstrate the performance difference that it implements.

Would happen in this code that calls gather(): https://github.com/simonw/asyncinject/blob/47348978242880bd72a444158bbecc64566b0c55/asyncinject/init.py#L114-L123
enhancement

opened by simonw 2
Ability to resolve an unregistered function
I'd like to be able to do the following:

async def one(): return 1 async def two(): return 2 registry = Registry(one, two) async def three(one, two): return one + two result = await registry.resolve(three)

Note that three has not been registered with the registry - but it still has its parameters inspected and used to resolve the dependencies.

This would be useful for Datasette, where I want plugins to be able to interact with predefined registries without needing to worry about picking a name for their function that doesn't clash with a name that has been registered by another plugin.
enhancement
opened by simonw 1
Try using __init_subclass__

https://twitter.com/dabeaz/status/1466731368956809219 - David Beazley says:

I think 95% of the problems once solved by a metaclass can be solved by __init_subclass__ instead

research

opened by simonw 1
Documentation needs a smarter example that illustrates graph dependencies

The examples in the README are boring, and don't show how the library can resolve a dependency tree into the most efficient possible mechanism.

Need to come up with a realistic example that demonstrates that.
documentation

opened by simonw 0

Releases(0.5)

0.5(Apr 22, 2022)

registry.resolve() can now be used to resolve functions that have not been registered. #13

async def one():
    return 1

async def two():
    return 2

registry = Registry(one, two)

async def three(one, two):
    return one + two

result = await registry.resolve(three)
# result is now 3

Source code(tar.gz)
Source code(zip)

0.4(Apr 18, 2022)
Adrian Garcia Badaracco addressed an issue where concurrent functions were not executing optimally. The code now starts a function running as soon as all of its dependent functions have completed. #10

Dropped support for Python 3.6.

Source code(tar.gz)
Source code(zip)
0.3(Apr 16, 2022)
Extensive, backwards-compatibility breaking redesign.

This library no longer uses subclasses. Instead, a Registry() object is created and async def functions are registered with that registry. The registry.resolve(fn) method is then used to execute functions with their dependencies. #8

Registry(timer=callable) can now be used to register a function to record the times taken to execute each function. This callable will be passed three arguments - the function name, the start time and the end time. #7

The parallel=True argument to the Registry() constructor can be switched to False to disable parallel execution - useful for running benchmarks to understand the performance benefit of running functions in parallel. #6

Source code(tar.gz)
Source code(zip)
0.2(Dec 21, 2021)
New _log mechanism for debug logging. #5

Source code(tar.gz)
Source code(zip)
0.2a1(Dec 3, 2021)
Now uses __init_subclass__ in place of a metaclass. #2

Source code(tar.gz)
Source code(zip)
0.2a0(Nov 17, 2021)
Provided parameters are now forwarded on to dependent methods.

Parameters with default values specified in the method signature are no longer treated as dependency injection parameters. #1

Source code(tar.gz)
Source code(zip)
0.1a0(Nov 17, 2021)
Initial alpha.

Source code(tar.gz)
Source code(zip)