background-task

Background Tasks ({{ cookiecutter.background_tasks }})

Tasks live in backend/app/worker/tasks/ (e.g. email_tasks.py, rag_tasks.py, cleanup_tasks.py). The app uses {{ cookiecutter.background_tasks }}. An in-process fallback (worker/background/) exists for trivial cases.

When to use a task vs. inline

Task: anything slow, retryable, or fire-and-forget — sending email, ingesting/embedding documents, calling slow external APIs, periodic cleanups, materialized-view refreshes.
Inline: fast, transactional work that the response depends on.

Add a task

Define it in backend/app/worker/tasks/<area>.py: {%- if cookiecutter.use_celery %}
```
from app.worker.celery_app import celery_app
```

@celery_app.task(name="send_welcome_email") def send_welcome_email(user_id: str) -> dict: ...

Enqueue: `send_welcome_email.delay(user_id)` (or `.apply_async(args=[...], countdown=60)`).
{%- elif cookiecutter.use_taskiq %}
```python
from app.worker.taskiq_app import broker

@broker.task
async def send_welcome_email(user_id: str) -> dict: ...

Enqueue: await send_welcome_email.kiq(user_id). {%- elif cookiecutter.use_arq %}

# add the coroutine, then list it in arq_app.WorkerSettings.functions
async def send_welcome_email(ctx, user_id: str) -> dict: ...

Enqueue: await request.state.arq_pool.enqueue_job("send_welcome_email", user_id). {%- elif cookiecutter.use_prefect %}

from prefect import flow

@flow(name="send-welcome-email", log_prints=True)
async def send_welcome_email_flow(user_id: str) -> dict: ...

Fire-and-forget from a service: asyncio.create_task(send_welcome_email_flow(user_id)). {%- endif %}

Call it from a service (not from the route directly) — keep business logic in services/, enqueue at the end of the unit of work.
Schedule it (optional): {%- if cookiecutter.use_celery %} add to beat_schedule in celery_app.py (run make celery-beat). {%- elif cookiecutter.use_taskiq %} append to SCHEDULES in tasks/schedules.py (run make taskiq-scheduler). {%- elif cookiecutter.use_arq %} add a cron(...) entry to WorkerSettings.cron_jobs in arq_app.py. {%- elif cookiecutter.use_prefect %} register a deployment with a CronSchedule/IntervalSchedule in app/worker/prefect_app.py. {%- endif %}
Run / verify: {%- if cookiecutter.use_celery %} make celery-worker (+ make celery-beat for schedules, make celery-flower to monitor). {%- elif cookiecutter.use_taskiq %} make taskiq-worker (+ make taskiq-scheduler for schedules). {%- elif cookiecutter.use_arq %} the worker runs in the dev stack (make dev), or uv run --directory backend arq app.worker.arq_app.WorkerSettings. {%- elif cookiecutter.use_prefect %} the prefect-server + prefect-runner start with make dev; watch runs at http://localhost:4200. {%- endif %}

Rules

Tasks take serializable args (ids, primitives) — not ORM objects or sessions. Re-fetch inside the task with a fresh session.
Make tasks idempotent where possible (safe to retry).
Keep heavy imports inside the task function to keep the API import-light.
See docs/howto/add-background-task.md for the full walkthrough.

About this skill

When to use a task vs. inline

Add a task

Rules