Recipe: Scheduled Reports¶
This recipe generates recurring reports on a cron schedule with @soniq.periodic. Soniq stores the schedule in PostgreSQL, and the soniq scheduler process enqueues each run.
The job¶
from soniq import Soniq
soniq = Soniq(database_url="postgresql://localhost/myapp")
@soniq.periodic(cron="0 9 * * 1", name="reports.weekly_sales_report") # Every Monday at 9 AM
async def weekly_sales_report():
start, end = get_last_week_range()
data = await fetch_sales_data(start, end)
report = generate_report(data)
await send_report(
to="team@example.com",
subject=f"Sales report: {start:%b %d} - {end:%b %d}",
attachment=report,
)
The @soniq.periodic decorator registers both the job and its schedule. The scheduler process reads the registered schedules.
Running the scheduler¶
The scheduler is a separate process that enqueues jobs on schedule. Run it next to your worker:
# Terminal 1: Scheduler (creates jobs on schedule)
SONIQ_JOBS_MODULES="app.reports" soniq scheduler
# Terminal 2: Worker (processes created jobs)
SONIQ_JOBS_MODULES="app.reports" soniq worker
The scheduler checks registered periodic jobs and enqueues them when their schedule fires. The worker processes them like any other job.
Schedule options¶
Cron expressions for specific times:
@soniq.periodic(cron="0 9 * * *", name="reports.daily_digest") # Daily at 9 AM
async def daily_digest():
...
@soniq.periodic(cron="0 0 1 * *", name="reports.monthly_summary") # First of every month
async def monthly_summary():
...
@soniq.periodic(cron="*/15 * * * *", name="orders.check_stale_orders") # Every 15 minutes
async def check_stale_orders():
...
Use the every(...) builders for simple intervals. They return cron strings:
from datetime import timedelta
from soniq import every
@soniq.periodic(cron=every(10).minutes(), name="maintenance.cleanup_temp_files")
async def cleanup_temp_files():
...
@soniq.periodic(cron=every(1).hours(), name="inventory.sync_inventory")
async def sync_inventory():
...
# Sub-minute uses every= directly (cron has no second resolution).
@soniq.periodic(every=timedelta(seconds=30), name="health.health_ping")
async def health_ping():
...
You can combine @periodic with any @job option:
@soniq.periodic(cron="0 9 * * 1", name="reports.weekly_report", queue="reports", retries=2, timeout=300)
async def weekly_report():
...
Complete example¶
# app/reports.py
from soniq import Soniq
soniq = Soniq(database_url="postgresql://localhost/myapp")
@soniq.periodic(cron="0 9 * * 1", name="reports.weekly_sales_report", queue="reports", timeout=300)
async def weekly_sales_report():
start, end = get_last_week_range()
data = await fetch_sales_data(start, end)
report = generate_csv(data)
await upload_to_s3(report, key=f"reports/sales-{start:%Y%m%d}.csv")
await send_email(
to="team@example.com",
subject=f"Weekly sales: {start:%b %d} - {end:%b %d}",
attachment_url=report.url,
)
@soniq.periodic(cron="0 6 * * *", name="reports.daily_error_digest", queue="reports")
async def daily_error_digest():
errors = await fetch_errors_since_yesterday()
if not errors:
return # Nothing to report
await send_email(
to="oncall@example.com",
subject=f"Error digest: {len(errors)} errors",
body=format_error_summary(errors),
)