Monitor Python Cron Jobs with Crontiq
Python scripts triggered by crontab, Airflow, or systemd timers are the backbone of data pipelines everywhere. They also fail silently. A cron entry like 0 3 * * * /usr/bin/python3 /opt/scripts/sync.py runs in the background. If it crashes, you get a local mail on the server that nobody reads. If the server reboots and crontab is wiped, the script simply stops running.
Crontiq adds dead-simple monitoring to any Python script. Add a few lines with the requests library, and you get alerts when the script misses its window, fails, or produces abnormal metrics.
Basic: Ping on Success
The simplest integration: call Crontiq at the end of your script. If the script crashes before reaching the ping, Crontiq marks the monitor as down after the expected interval.
#!/usr/bin/env python3
"""sync_orders.py — Sync orders from API to database."""
import requests
from app.orders import sync_all_orders
CRONTIQ_URL = "https://ping.crontiq.io/p/cq_live_80381902d7b36613/order-sync"
def main():
count = sync_all_orders()
print(f"Synced {count} orders")
# Ping Crontiq on success
requests.get(CRONTIQ_URL, timeout=5)
if __name__ == "__main__":
main()
The monitor order-sync is created automatically on the first ping. No dashboard configuration needed. Crontiq learns the schedule from ping frequency.
Production Pattern: Start, Metrics, and Failure Handling
For production scripts, signal the start of execution, report metrics on success, and explicitly report failures. This gives you duration tracking and anomaly detection.
#!/usr/bin/env python3
"""etl_pipeline.py — Nightly ETL from source DB to warehouse."""
import os
import time
import requests
API_KEY = os.environ["CRONTIQ_API_KEY"]
BASE_URL = f"https://ping.crontiq.io/p/{API_KEY}/nightly-etl"
def ping(action="", json_body=None):
"""Send a ping to Crontiq. Never let monitoring break the job."""
try:
url = f"{BASE_URL}/{action}" if action else BASE_URL
if json_body:
requests.post(url, json=json_body, timeout=5)
else:
requests.get(url, timeout=5)
except requests.RequestException:
pass # Monitoring must never break the job
def main():
ping("start")
start_time = time.time()
try:
# Your actual ETL logic
rows = extract_from_source()
transformed = transform(rows)
loaded = load_to_warehouse(transformed)
duration_ms = int((time.time() - start_time) * 1000)
# Report success with metrics
ping(json_body={
"rows_extracted": len(rows),
"rows_loaded": loaded,
"duration_ms": duration_ms,
"transform_errors": len(rows) - len(transformed),
})
except Exception as e:
ping("fail")
raise
if __name__ == "__main__":
main()
Context Manager Pattern
If you run many Python cron scripts, wrap the monitoring logic in a reusable context manager. This keeps each script clean and consistent.
import os
import time
import requests
from contextlib import contextmanager
@contextmanager
def crontiq(slug):
"""Context manager for Crontiq monitoring."""
api_key = os.environ.get("CRONTIQ_API_KEY", "")
base = f"https://ping.crontiq.io/p/{api_key}/{slug}"
metrics = {}
def safe_ping(url, json_body=None):
try:
if json_body:
requests.post(url, json=json_body, timeout=5)
else:
requests.get(url, timeout=5)
except Exception:
pass
safe_ping(f"{base}/start")
start = time.time()
try:
yield metrics
metrics["duration_ms"] = int((time.time() - start) * 1000)
safe_ping(base, json_body=metrics)
except Exception:
safe_ping(f"{base}/fail")
raise
# Usage in any script:
def main():
with crontiq("report-generator") as m:
reports = generate_reports()
m["reports_generated"] = len(reports)
m["total_pages"] = sum(r.page_count for r in reports)
Sending Detailed Metrics
Crontiq flattens nested JSON automatically. You can send any structure your script produces — no schema configuration required.
# Your script sends this:
requests.post(CRONTIQ_URL, json={
"database": {
"rows_inserted": 4521,
"rows_updated": 128,
"rows_deleted": 3,
},
"api": {
"requests_made": 45,
"errors": 0,
"avg_latency_ms": 220,
},
"duration_sec": 34,
})
# Crontiq tracks these metrics:
# database.rows_inserted = 4521
# database.rows_updated = 128
# database.rows_deleted = 3
# api.requests_made = 45
# api.errors = 0
# api.avg_latency_ms = 220
# duration_sec = 34
Crontab Entry
Set the CRONTIQ_API_KEY environment variable in your crontab so all scripts can access it:
# /etc/crontab or crontab -e
CRONTIQ_API_KEY=cq_live_80381902d7b36613
0 3 * * * /usr/bin/python3 /opt/scripts/etl_pipeline.py 2>>/var/log/etl.log
30 * * * * /usr/bin/python3 /opt/scripts/sync_orders.py 2>>/var/log/sync.log
Why Crontiq Instead of Logging?
- Logs tell you what happened if you read them. Crontiq tells you when something did not happen.
- A script that never runs produces no logs. Crontiq detects the absence.
- Crontiq tracks numeric trends and flags anomalies: a sync job that usually processes 10,000 rows but returns 12 is suspicious even if it exits cleanly.
- One dashboard for all your Python scripts across all servers.