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

1. 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 %}

   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 %}

2. Call it from a service (not from the route directly) — keep business logic in services/, enqueue at the end of the unit of work.

3. 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 %}