Celery Beat Crontab Schedules

from celery import Celery
from celery.schedules import crontab

app = Celery('ops', broker='redis://localhost:6379/1')
app.conf.update(
    timezone='UTC',
    beat_schedule={
        'nightly-invoice-close': {
            'task': 'tasks.close_invoices',
            'schedule': crontab(hour=2, minute=0),
            'options': {'queue': 'billing'},
        },
        'cache-warm-every-15min': {
            'task': 'tasks.warm_cache',
            'schedule': crontab(minute='*/15'),
            'options': {'queue': 'maintenance'},
        },
        'staggered-region-reports': {
            'task': 'tasks.region_report',
            'schedule': crontab(hour=6, minute=30, day_of_week='mon'),
            'args': ('eu',),
            'options': {'queue': 'reporting'},
        },
        'monthly-compliance-export': {
            'task': 'tasks.compliance_export',
            'schedule': crontab(day=1, hour=3, minute=0),
            'options': {'queue': 'reporting', 'expires': 7200},
        },
    },
)


@app.task(queue='billing')
def close_invoices() -> dict:
    return {'closed': 42}


@app.task(queue='maintenance')
def warm_cache() -> dict:
    return {'warmed': ['rates', 'catalog']}


@app.task(queue='reporting')
def region_report(region: str) -> dict:
    return {'region': region, 'rows': 128}


@app.task(queue='reporting')
def compliance_export() -> dict:
    return {'archive': 'compliance-2026-10.zip'}