ManagedBackgroundTasks

A BackgroundTasks subclass that intercepts add_task to inject retries, status tracking, and task IDs.

Because it inherits from BackgroundTasks, isinstance checks pass and it is a drop-in replacement.

Constructor

ManagedBackgroundTasks(
    task_manager: TaskManager,
    background_tasks: BackgroundTasks | None = None,
)

When constructed via a dependency or install(), FastAPI passes its native BackgroundTasks instance. The wrapper shares the native task list so Starlette executes the wrapped tasks after the response is sent.

Methods

add_task(func, *args, idempotency_key=None, tags=None, **kwargs) -> str

Enqueues a function as a managed background task. Returns the task_id UUID string.

This overrides BackgroundTasks.add_task which returns None. If you are capturing the return value in existing code, this is the only behavioural change.

task_id = background_tasks.add_task(send_email, address=email)

Parameters:

Parameter	Type	Default	Description
func	Callable	required	The task function to run. Must be registered with @task_manager.task() to get retries and config applied.
*args			Positional arguments forwarded to func.
idempotency_key	str \| None	None	Deduplication key. If a non-failed task with the same key already exists, its task_id is returned and func is not enqueued again.
tags	dict[str, str] \| None	None	Key/value labels attached to this task. Forwarded to every LogEvent and LifecycleEvent.
**kwargs			Keyword arguments forwarded to func.

Direct construction

You can construct ManagedBackgroundTasks directly outside of a request context, for example in tests:

from fastapi_taskflow import ManagedBackgroundTasks, TaskManager

tm = TaskManager()
managed = ManagedBackgroundTasks(tm)
task_id = managed.add_task(my_func, arg)

Tasks added this way will not be executed by Starlette automatically. You would need to call them manually or run them inside a request context.