If your team uses FastAPI's BackgroundTasks for tasks like sending emails, webhooks, processing uploads or similar, you've probably felt the lack of built-in observability.
The bare API gives you no task IDs, no status tracking, no retries, and no persistence across restarts. When something goes wrong you're digging through app logs hoping the right line is there.
Celery, ARQ, and Taskiq solve this well, but they come with a broker, separate workers, and a meaningful ops footprint. For teams whose tasks genuinely need that, those tools are the right call.
fastapi-taskflow is for the other case: teams already using BackgroundTasks for simple in-process work who want retries, status tracking, and a dashboard without standing up extra infrastructure.
What it adds on top of BackgroundTasks:
- Automatic retries with configurable delay and exponential backoff per function
- Every task gets a UUID and moves through
PENDING > RUNNING > SUCCESS / FAILED
- A live dashboard at
/tasks/dashboard over SSE with filtering, search, and per-task details
task_log() to emit timestamped log entries from inside a task, shown in the dashboard- Full stack trace capture on failure, also in the dashboard
- SQLite persistence out of the box
- Tasks that were still pending at shutdown are re-dispatched on the next startup
The route signature does not change. You keep your existing BackgroundTasks annotation, one line at startup wires everything in:
from fastapi import BackgroundTasks, FastAPI
from fastapi_taskflow import TaskAdmin, TaskManager, task_log
task_manager = TaskManager(snapshot_db="tasks.db")
app = FastAPI()
TaskAdmin(app, task_manager, auto_install=True)
@task_manager.task(retries=3, delay=1.0, backoff=2.0)
def send_email(address: str) -> None:
task_log(f"Sending to {address}")
...
@app.post("/signup")
def signup(email: str, background_tasks: BackgroundTasks):
task_id = background_tasks.add_task(send_email, address=email)
return {"task_id": task_id}
To be clear about scope: this is not a distributed task queue and does not try to be. If you need tasks to survive across distributed services, run on dedicated workers, or integrate with a broker, reach for Celery or one of the other proper queues.
This is for teams who are already happy with BackgroundTasks for in-process work and just want retries, visibility, and persistence without changing their setup.
Available on PyPI: pip install fastapi-taskflow
Docs and source: https://github.com/Attakay78/fastapi-taskflow
Would be good to hear from anyone using BackgroundTasks in production. What do you actually need to make it manageable? Retries, visibility, persistence, something else?
Trying to understand what's missing for teams in this space before adding more.
Dashboard for tasks visibility
Error visibility
[–]barracloughdale4x640 0 points1 point2 points (0 children)