Dependency Injection

Hatchet’s Python SDK allows you to inject dependencies into your tasks, FastAPI style. These dependencies can be either synchronous or asynchronous functions. They are executed before the task is triggered, and their results are injected into the task as parameters.

This behaves almost identically to FastAPI’s dependency injection, and is intended to be used in the same way. Dependencies are useful for sharing logic between tasks that you’d like to avoid repeating, or would like to factor out of the task logic itself (e.g. to make testing easier).

Usage

To add dependencies to your tasks, import Depends from the hatchet_sdk. Then:

async def async_dep(input: EmptyModel, ctx: Context) -> str:
    return ASYNC_DEPENDENCY_VALUE


def sync_dep(input: EmptyModel, ctx: Context) -> str:
    return SYNC_DEPENDENCY_VALUE

In this example, we’ve declared two dependencies: one synchronous and one asynchronous. You can do anything you like in your dependencies, such as creating database sessions, managing configuration, sharing instances of service-layer logic, and more.

Once you’ve defined your dependency functions, inject them into your tasks as follows:

@hatchet.task()
async def async_task_with_dependencies(
    _i: EmptyModel,
    ctx: Context,
    async_dep: Annotated[str, Depends(async_dep)],
    sync_dep: Annotated[str, Depends(sync_dep)],
) -> Output:
    return Output(
        sync_dep=sync_dep,
        async_dep=async_dep,
    )

That’s it! Now, whenever your task is triggered, its dependencies will be evaluated, and the results will be injected into the task at runtime for you to use as needed.