Run scheduled tasks alongside FastAPI

Approach

Don't run schedulers inside every web worker — with 4 workers × 3 replicas the job runs 12 times. Run scheduling in one dedicated process (ARQ/Celery beat cron jobs, a Kubernetes CronJob, or systemd timers), and make jobs idempotent with a distributed lock as a safety net.

Solution

# ARQ cron jobs in the worker process (not in the API)
from arq import cron


async def purge_expired_sessions(ctx) -> None:
    async with SessionLocal() as db:
        await db.execute(delete(Session).where(Session.expires_at < utcnow()))
        await db.commit()


async def nightly_report(ctx) -> None:
    lock = ctx["redis"].lock("lock:nightly_report", timeout=3600)
    if not await lock.acquire(blocking=False):
        return                                      # another worker is running it
    try:
        await build_and_email_report()
    finally:
        await lock.release()


class WorkerSettings:
    functions = []
    cron_jobs = [
        cron(purge_expired_sessions, minute=0),             # hourly
        cron(nightly_report, hour=2, minute=15),           # 02:15 UTC
    ]
    redis_settings = RedisSettings.from_dsn("redis://redis:6379/1")

# Alternative: Kubernetes CronJob reusing the API image
apiVersion: batch/v1
kind: CronJob
metadata: { name: nightly-report }
spec:
  schedule: "15 2 * * *"
  concurrencyPolicy: Forbid
  jobTemplate:
    spec:
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: report
              image: registry.example.com/api:1.42.0
              command: ["python", "-m", "app.jobs.nightly_report"]

Pitfalls

• APScheduler started in the FastAPI lifespan of every worker.
• Jobs that aren't idempotent when retried.
• No alert when a scheduled job silently stops running.

Variations

• Celery beat with a single beat process.
• Use a heartbeat monitor (healthchecks.io) per job.

Further reading: ARQ: Cron jobs