Skip to content

Commit

Permalink
Implemented external API to run functions with dependency being injec…
Browse files Browse the repository at this point in the history
…ted (#12)

* Add external API on aioclock to run any task with deps injected

* Add test
  • Loading branch information
ManiMozaffar authored May 23, 2024
1 parent 0cf0c91 commit 9d4edf0
Show file tree
Hide file tree
Showing 3 changed files with 50 additions and 1 deletion.
3 changes: 2 additions & 1 deletion aioclock/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from fast_depends import Depends

from .app import AioClock
from .app import AioClock, run_with_injected_deps
from .group import Group
from .triggers import At, Every, Forever, Once, OnShutDown, OnStartUp

Expand All @@ -14,4 +14,5 @@
"Group",
"AioClock",
"At",
"run_with_injected_deps",
]
27 changes: 27 additions & 0 deletions aioclock/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -178,3 +178,30 @@ async def serve(self) -> None:
finally:
shutdown_tasks = self._get_shutdown_task()
await asyncio.gather(*(task.run() for task in shutdown_tasks), return_exceptions=False)


async def run_with_injected_deps(func: Callable[P, Awaitable[T]]) -> T:
"""Runs an aioclock decorated function, with all the dependencies injected.
Example:
```python
from aioclock import run_with_injected_deps, Once, AioClock, Depends
app = AioClock()
def some_dependency():
return 1
@app.task(trigger=Once())
async def main(bar: int = Depends(some_dependency)):
print("Hello World")
return bar
async def some_other_func():
foo = await run_with_injected_deps(main)
assert foo == 1
```
"""
return await inject(func, dependency_overrides_provider=get_provider())() # type: ignore
21 changes: 21 additions & 0 deletions tests/test_di.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
import pytest

from aioclock import AioClock, Depends, Once, run_with_injected_deps

app = AioClock()


def some_dependency():
return 1


@app.task(trigger=Once())
async def main(bar: int = Depends(some_dependency)):
print("Hello World")
return bar


@pytest.mark.asyncio
async def test_run_manual():
foo = await run_with_injected_deps(main)
assert foo == 1

0 comments on commit 9d4edf0

Please sign in to comment.