Skip to content

FastAPI-TaskFlow

Backends

fastapi-taskflow

Backends¶

SnapshotBackend (ABC)¶

from fastapi_taskflow.backends.base import SnapshotBackend

Abstract base class all backends must implement. See Custom Backends for the full method signatures and a worked example.

SqliteBackend¶

from fastapi_taskflow.backends.sqlite import SqliteBackend

backend = SqliteBackend("tasks.db")

Stores history in a task_snapshots table and pending tasks in a task_pending_requeue table. Handles schema migrations automatically on first connection.

Extra method: `query()`¶

Not part of the ABC. Only available on SqliteBackend.

records = backend.query(
    status: str | None = None,
    func_name: str | None = None,
    limit: int = 100,
) -> list[dict]

Or via the scheduler shortcut:

task_manager._scheduler.query(status="failed", func_name="send_email")

RedisBackend¶

pip install "fastapi-taskflow[redis]"

from fastapi_taskflow.backends.redis import RedisBackend

backend = RedisBackend(
    url: str = "redis://localhost:6379/0",
    prefix: str = "fastapi_taskflow",
)

Stores history as Redis hashes with a set-based index. Pending tasks are stored in a separate key namespace.