| Name | asyncinject JSON |
| Version |
0.6.1
JSON |
| download |
| home_page | None |
| Summary | Run async workflows using pytest-fixtures-style dependency injection |
| upload_time | 2025-10-09 03:23:11 |
| maintainer | None |
| docs_url | None |
| author | Simon Willison |
| requires_python | >=3.10 |
| license | None |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# asyncinject
[](https://pypi.org/project/asyncinject/)
[](https://github.com/simonw/asyncinject/releases)
[](https://github.com/simonw/asyncinject/blob/main/LICENSE)
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](https://docs.pytest.org/en/6.2.x/fixture.html).
The idea is to simplify executing parallel `asyncio` operations by allowing them to be defined using a collection of functions, where the function arguments represent dependent functions that need to be executed first.
The library can then create and execute a plan for executing the required functions in parallel in the most efficient sequence possible.
Here's an example, using the [httpx](https://www.python-httpx.org/) HTTP library.
```python
from asyncinject import Registry
import httpx
async def get(url):
async with httpx.AsyncClient() as client:
return (await client.get(url)).text
async def example():
return await get("http://www.example.com/")
async def simonwillison():
return await get("https://simonwillison.net/search/?tag=empty")
async def both(example, simonwillison):
return example + "\n\n" + simonwillison
registry = Registry(example, simonwillison, both)
combined = await registry.resolve(both)
print(combined)
```
If you run this in `ipython` or `python -m asyncio` (to enable top-level await in the console) 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 notices that `both()` takes two arguments which are the names of other registered `async def` functions, and will construct an execution plan that executes those two functions in parallel, then passes their results to the `both()` method.
### Registry.from_dict()
Passing a list of functions to the `Registry` constructor will register each function under their introspected function name, using `fn.__name__`.
You can set explicit names instead using a dictionary:
```python
registry = Registry.from_dict({
"example": example,
"simonwillison": simonwillison,
"both": both
})
```
Those string names will be used to match parameters, so each function will need to accept parameters named after the keys used in that dictionary.
### Registering additional functions
Functions that are registered can be regular functions or `async def` functions.
In addition to registering functions by passing them to the constructor, you can also add them to a registry using the `.register()` method:
```python
async def another():
return "another"
registry.register(another)
```
To register them with a name other than the name of the function, pass the `name=` argument:
```python
async def another():
return "another 2"
registry.register(another, name="another_2")
```
### Resolving an unregistered function
You don't need to register the final function that you pass to `.resolve()` - if you pass an unregistered function, the library will introspect the function's parameters and resolve them directly.
This works with both regular and async functions:
```python
async def one():
return 1
async def two():
return 2
registry = Registry(one, two)
# async def works here too:
def three(one, two):
return one + two
print(await registry.resolve(three))
# Prints 3
```
### Parameters are passed through
Your dependent functions can require keyword arguments which have been passed to the `.resolve()` call:
```python
async def get_param_1(param1):
return await get(param1)
async def get_param_2(param2):
return await get(param2)
async def both(get_param_1, get_param_2):
return get_param_1 + "\n\n" + get_param_2
combined = await Registry(get_param_1, get_param_2, both).resolve(
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:
```python
async def go(calc1, x=5):
return calc1 + x
async def calc1():
return 5
print(await Registry(calc1, go).resolve(go))
# Prints 10
```
### Tracking with a timer
You can pass a `timer=` callable to the `Registry` constructor to gather timing information about executed tasks.. Your function should take three positional arguments:
- `name` - the name of the function that is being timed
- `start` - the time that it started executing, using `time.perf_counter()` ([perf_counter() docs](https://docs.python.org/3/library/time.html#time.perf_counter))
- `end` - the time that it finished executing
You can use `print` here too:
```python
combined = await Registry(
get_param_1, get_param_2, both, timer=print
).resolve(
both,
param1 = "http://www.example.com/",
param2 = "https://simonwillison.net/search/?tag=empty"
)
```
This will output:
```
get_param_1 436633.584580685 436633.797921747
get_param_2 436633.641832699 436634.196364347
both 436634.196570217 436634.196575639
```
### Turning off parallel execution
By default, functions that can run in parallel according to the execution plan will run in parallel using `asyncio.gather()`.
You can disable this parallel exection by passing `parallel=False` to the `Registry` constructor, or by setting `registry.parallel = False` after the registry object has been created.
This is mainly useful for benchmarking the difference between parallel and serial execution for your project.
## 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
Now install the dependencies and test dependencies:
pip install -e '.[test]'
To run the tests:
pytest
Raw data
{
"_id": null,
"home_page": null,
"name": "asyncinject",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": null,
"author": "Simon Willison",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/6f/4e/62c1c1f4a63ac3b6d251b170be57dc8d463b93d23da95de70bea51615b25/asyncinject-0.6.1.tar.gz",
"platform": null,
"description": "# asyncinject\n\n[](https://pypi.org/project/asyncinject/)\n[](https://github.com/simonw/asyncinject/releases)\n[](https://github.com/simonw/asyncinject/blob/main/LICENSE)\n\nRun async workflows using pytest-fixtures-style dependency injection\n\n## Installation\n\nInstall this library using `pip`:\n\n $ pip install asyncinject\n\n## Usage\n\nThis library is inspired by [pytest fixtures](https://docs.pytest.org/en/6.2.x/fixture.html).\n\nThe idea is to simplify executing parallel `asyncio` operations by allowing them to be defined using a collection of functions, where the function arguments represent dependent functions that need to be executed first.\n\nThe library can then create and execute a plan for executing the required functions in parallel in the most efficient sequence possible.\n\nHere's an example, using the [httpx](https://www.python-httpx.org/) HTTP library.\n\n```python\nfrom asyncinject import Registry\nimport httpx\n\n\nasync def get(url):\n async with httpx.AsyncClient() as client:\n return (await client.get(url)).text\n\nasync def example():\n return await get(\"http://www.example.com/\")\n\nasync def simonwillison():\n return await get(\"https://simonwillison.net/search/?tag=empty\")\n\nasync def both(example, simonwillison):\n return example + \"\\n\\n\" + simonwillison\n\nregistry = Registry(example, simonwillison, both)\ncombined = await registry.resolve(both)\nprint(combined)\n```\nIf you run this in `ipython` or `python -m asyncio` (to enable top-level await in the console) you will see output that combines HTML from both of those pages.\n\nThe HTTP requests to `www.example.com` and `simonwillison.net` will be performed in parallel.\n\nThe library notices that `both()` takes two arguments which are the names of other registered `async def` functions, and will construct an execution plan that executes those two functions in parallel, then passes their results to the `both()` method.\n\n### Registry.from_dict()\n\nPassing a list of functions to the `Registry` constructor will register each function under their introspected function name, using `fn.__name__`.\n\nYou can set explicit names instead using a dictionary:\n\n```python\nregistry = Registry.from_dict({\n \"example\": example,\n \"simonwillison\": simonwillison,\n \"both\": both\n})\n```\nThose string names will be used to match parameters, so each function will need to accept parameters named after the keys used in that dictionary.\n\n### Registering additional functions\n\nFunctions that are registered can be regular functions or `async def` functions.\n\nIn addition to registering functions by passing them to the constructor, you can also add them to a registry using the `.register()` method:\n\n```python\nasync def another():\n return \"another\"\n\nregistry.register(another)\n```\nTo register them with a name other than the name of the function, pass the `name=` argument:\n```python\nasync def another():\n return \"another 2\"\n\nregistry.register(another, name=\"another_2\")\n```\n\n### Resolving an unregistered function\n\nYou don't need to register the final function that you pass to `.resolve()` - if you pass an unregistered function, the library will introspect the function's parameters and resolve them directly.\n\nThis works with both regular and async functions:\n\n```python\nasync def one():\n return 1\n\nasync def two():\n return 2\n\nregistry = Registry(one, two)\n\n# async def works here too:\ndef three(one, two):\n return one + two\n\nprint(await registry.resolve(three))\n# Prints 3\n```\n\n### Parameters are passed through\n\nYour dependent functions can require keyword arguments which have been passed to the `.resolve()` call:\n\n```python\nasync def get_param_1(param1):\n return await get(param1)\n\nasync def get_param_2(param2):\n return await get(param2)\n\nasync def both(get_param_1, get_param_2):\n return get_param_1 + \"\\n\\n\" + get_param_2\n\n\ncombined = await Registry(get_param_1, get_param_2, both).resolve(\n both,\n param1 = \"http://www.example.com/\",\n param2 = \"https://simonwillison.net/search/?tag=empty\"\n)\nprint(combined)\n```\n### Parameters with default values are ignored\n\nYou can opt a parameter out of the dependency injection mechanism by assigning it a default value:\n\n```python\nasync def go(calc1, x=5):\n return calc1 + x\n\nasync def calc1():\n return 5\n\nprint(await Registry(calc1, go).resolve(go))\n# Prints 10\n```\n\n### Tracking with a timer\n\nYou can pass a `timer=` callable to the `Registry` constructor to gather timing information about executed tasks.. Your function should take three positional arguments:\n\n- `name` - the name of the function that is being timed\n- `start` - the time that it started executing, using `time.perf_counter()` ([perf_counter() docs](https://docs.python.org/3/library/time.html#time.perf_counter))\n- `end` - the time that it finished executing\n\nYou can use `print` here too:\n\n```python\ncombined = await Registry(\n get_param_1, get_param_2, both, timer=print\n).resolve(\n both,\n param1 = \"http://www.example.com/\",\n param2 = \"https://simonwillison.net/search/?tag=empty\"\n)\n```\nThis will output:\n```\nget_param_1 436633.584580685 436633.797921747\nget_param_2 436633.641832699 436634.196364347\nboth 436634.196570217 436634.196575639\n```\n### Turning off parallel execution\n\nBy default, functions that can run in parallel according to the execution plan will run in parallel using `asyncio.gather()`.\n\nYou can disable this parallel exection by passing `parallel=False` to the `Registry` constructor, or by setting `registry.parallel = False` after the registry object has been created.\n\nThis is mainly useful for benchmarking the difference between parallel and serial execution for your project.\n\n## Development\n\nTo contribute to this library, first checkout the code. Then create a new virtual environment:\n\n cd asyncinject\n python -m venv venv\n source venv/bin/activate\n\nNow install the dependencies and test dependencies:\n\n pip install -e '.[test]'\n\nTo run the tests:\n\n pytest\n",
"bugtrack_url": null,
"license": null,
"summary": "Run async workflows using pytest-fixtures-style dependency injection",
"version": "0.6.1",
"project_urls": {
"CI": "https://github.com/simonw/asyncinject/actions",
"Changelog": "https://github.com/simonw/asyncinject/releases",
"Homepage": "https://github.com/simonw/asyncinject",
"Issues": "https://github.com/simonw/asyncinject/issues"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "9246a669bc58f829dd0b873ad570c22342cd8d34ef17a8e203e320efa3df7c5d",
"md5": "9aa22e4ddd928028279d347b46b38f97",
"sha256": "3723128f8016b55a700cf89839624ef1b3b42c0c24a94b3be921326087aa2cc7"
},
"downloads": -1,
"filename": "asyncinject-0.6.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9aa22e4ddd928028279d347b46b38f97",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 12392,
"upload_time": "2025-10-09T03:23:09",
"upload_time_iso_8601": "2025-10-09T03:23:09.762525Z",
"url": "https://files.pythonhosted.org/packages/92/46/a669bc58f829dd0b873ad570c22342cd8d34ef17a8e203e320efa3df7c5d/asyncinject-0.6.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "6f4e62c1c1f4a63ac3b6d251b170be57dc8d463b93d23da95de70bea51615b25",
"md5": "acb54c2c27739c2dca17c928e853870c",
"sha256": "7f9a08e7bbbc45d901ee620351f486a140ef8ca6bf4c1ef6e1e28309a0a5e5b8"
},
"downloads": -1,
"filename": "asyncinject-0.6.1.tar.gz",
"has_sig": false,
"md5_digest": "acb54c2c27739c2dca17c928e853870c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 13445,
"upload_time": "2025-10-09T03:23:11",
"upload_time_iso_8601": "2025-10-09T03:23:11.098196Z",
"url": "https://files.pythonhosted.org/packages/6f/4e/62c1c1f4a63ac3b6d251b170be57dc8d463b93d23da95de70bea51615b25/asyncinject-0.6.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-10-09 03:23:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "simonw",
"github_project": "asyncinject",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "asyncinject"
}
