Celery Beat scheduling (with FastAPI)

Beat (single) ──schedule tick──► RabbitMQ ──► workers ──► side effects / job rows

┌─────────────┐ schedule ┌──────────┐ tasks ┌──────────┐
│ Celery Beat │ ────────────────► │ RabbitMQ │ ─────────────► │ workers │
│ (1 leader) │ │ jobs.* │ │ N pods │
└─────────────┘ └──────────┘ └──────────┘
 │ │
 │ never shares process with API │
 ▼ ▼
 separate container/service updates jobs / DB

# worker
celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io -c 4

# beat (single replica)
celery -A app.workers.celery_app.celery_app beat -l info

# every 300 seconds
"sync-vendor": {
 "task": "app.workers.tasks.sync.pull_vendor",
 "schedule": 300.0,
}

from celery.schedules import crontab

"nightly-settle": {
 "task": "app.workers.tasks.billing.settle_day",
 "schedule": crontab(hour=2, minute=15),
 "options": {"queue": "jobs.heavy"},
}

# PSEUDOCODE : app/workers/celery_app.py
from celery import Celery
from celery.schedules import crontab
from app.core.settings import settings

celery_app = Celery("app", broker=settings.celery_broker_url)

celery_app.conf.update(
 timezone="UTC", # store and think in UTC
 enable_utc=True,
 task_serializer="json",
 accept_content=["json"],
 beat_schedule={
 "reconcile-outbox-every-minute": {
 "task": "app.workers.tasks.outbox.relay_tick",
 "schedule": 60.0,
 "options": {"queue": "jobs.default"},
 },
 "daily-prune-jobs": {
 "task": "app.workers.tasks.maintenance.prune_old_jobs",
 "schedule": crontab(hour=3, minute=0), # 03:00 UTC
 "options": {"queue": "jobs.default"},
 },
 "weekday-report": {
 "task": "app.workers.tasks.reports.daily_summary",
 "schedule": crontab(hour=8, minute=0, day_of_week="1-5"),
 "args": (), # prefer no fat args
 "options": {"queue": "jobs.io"},
 },
 },
 beat_max_loop_interval=5, # how often Beat wakes to check schedule
)

# PSEUDOCODE : settings
class Settings(BaseSettings):
 celery_broker_url: str
 celery_beat_enabled: bool = True # disable in most containers

# PSEUDOCODE : good: schedule enqueues unit of work
@celery_app.task(name="app.workers.tasks.billing.settle_day")
def settle_day():
 # optional: create a jobs row for audit/UI
 job_id = create_job(type="settle_day", status="pending")
 run_settle(job_id) # or split: only enqueue children

# PSEUDOCODE
@celery_app.task(name="billing.settle_day_fanout")
def settle_day_fanout():
 for merchant_id in list_merchants_due():
 settle_merchant.delay(merchant_id) # many small tasks

# PSEUDOCODE
@celery_app.task(name="reports.daily_summary")
def daily_summary():
 day = utc_today_iso()
 key = f"daily_summary:{day}"
 if not acquire_once(key, ttl=36 * 3600): # Redis lock or UNIQUE job key
 return # already ran
 generate_summary(day)

UNIQUE (type, period_key) -- type='daily_summary', period_key='2026-08-05'

# PSEUDOCODE k8s idea
# Deployment beat: replicas: 1
# PodDisruptionBudget optional; prefer schedule tolerance over multi-beat

Cloud scheduler / CronJob → POST /internal/schedules/daily-summary
 → enqueue task (same as Beat would)

# PSEUDOCODE : business local window computed inside task
@celery_app.task
def market_open_tick():
 if not is_market_open(zone="America/New_York"):
 return
 ...

# PSEUDOCODE : reconcilable periodic work
def settle_day(day: str | None = None):
 day = day or utc_yesterday()
 if already_settled(day):
 return
 do_settle(day)

# PSEUDOCODE : schedule every 60s
@celery_app.task(name="ops.beat_heartbeat")
def beat_heartbeat():
 metrics.gauge("beat_heartbeat_unixtime", time.time())

# PSEUDOCODE : tasks/maintenance.py
@celery_app.task(name="maintenance.prune_old_jobs")
def prune_old_jobs(days: int = 30):
 deleted = db.execute(
 "DELETE FROM jobs WHERE status IN ('succeeded','failed') "
 "AND finished_at < now() - interval '%s days'" % days
 ) # use bound params in real code
 log.info("pruned", extra={"deleted": deleted})

# celery_app.conf.beat_schedule entry points at this task daily

# production processes
celery -A app.workers.celery_app.celery_app worker -Q jobs.default,jobs.io,jobs.heavy
celery -A app.workers.celery_app.celery_app beat -l info