taskiq-dependencies

1.5.7 · active · verified Sun Apr 12

taskiq-dependencies is a Python library providing a FastAPI-like dependency injection system. It allows for defining dependencies as functions and automatically resolving them for other callables. As a core component of the Taskiq asynchronous task queue ecosystem, it is actively maintained with frequent updates, currently at version 1.5.7.

Warnings

gotcha Unlike web frameworks such as FastAPI, `taskiq-dependencies` does not automatically inject dependencies. You must explicitly call `inject_dependencies(your_function)` to resolve and execute the function with its dependencies.
Fix: Always wrap your dependent function call with `inject_dependencies()`.
gotcha Dependency lifecycle and scope are single-shot per `inject_dependencies` call. This library does not provide request-scoped, session-scoped, or singleton dependency management out-of-the-box like a full web framework might. Each call to `inject_dependencies` re-evaluates all dependencies.
Fix: Be mindful of the performance implications if dependencies are expensive to compute. For specific lifecycle management within `taskiq` tasks, consider how `taskiq`'s own integration (e.g., `TaskiqDepends`) might layer on top.
gotcha While `taskiq-dependencies.Depends` works directly, when using it within `taskiq` tasks, it's often more idiomatic and potentially safer to use `taskiq.TaskiqDepends` if available. `Taskiq` uses `taskiq-dependencies` under the hood and may provide additional integration points or overrides specific to the task execution environment.
Fix: For `taskiq` tasks, prefer `from taskiq import TaskiqDepends as Depends` to ensure full compatibility with the task queue's features and lifecycle management.

Install

pip install taskiq-dependencies Install stable version

Imports

Depends
```
from taskiq_dependencies import Depends
```

inject_dependencies

from taskiq_dependencies import inject_dependencies

Quickstart

This quickstart demonstrates how to define and use dependencies with `taskiq-dependencies`. It covers both synchronous and asynchronous dependencies, showing how to use `Depends` to declare requirements and `inject_dependencies` to manually resolve and call the target function.

from taskiq_dependencies import Depends, inject_dependencies

# A simple dependency that returns a user ID
def get_user_id() -> int:
    return 1

# A function that uses the dependency
def greet_user(user_id: int = Depends(get_user_id)) -> str:
    return f"Hello, user {user_id}"

# Manually inject dependencies and call the function
result = inject_dependencies(greet_user)
print(result)

# Example with async dependency
import asyncio

async def get_async_value() -> str:
    await asyncio.sleep(0.01)
    return "Async Value"

async def process_async_data(value: str = Depends(get_async_value)) -> str:
    return f"Processed: {value}"

# For async functions, inject_dependencies also returns an awaitable
async def main():
    async_result = await inject_dependencies(process_async_data)
    print(async_result)

if __name__ == '__main__':
    asyncio.run(main())

view raw JSON →