zuvio-atlas 0.1.0

Zuvio Atlas job/cron-run monitoring SDK — report start/finish/fail of scheduled jobs.

zuvio-atlas — Python job monitoring SDK

Report your cron scripts, Celery tasks, and Airflow jobs to Zuvio Atlas. Wrap a function and Atlas records each run — start, finish/fail, duration, and the exception on failure — then alerts you on failed, overrun, or missed runs.

Python 3.8+, standard library only (no dependencies).

Install

pip install zuvio-atlas

Use

import zuvio_atlas as atlas

# The slug comes from the monitor's page (Uptime → Cron Jobs) in the dashboard.
@atlas.job("abc123")
def run_backup():
    ...

As a context manager:

with atlas.monitor("abc123"):
    run_backup()

Manual control:

run_id = atlas.start("abc123")
try:
    do_work()
    atlas.finish("abc123", run_id=run_id, metrics={"rows": 1200})
except Exception as e:
    atlas.fail("abc123", run_id=run_id, message=str(e))
    raise

API-key mode (monitoring-as-code)

Set a pm_live_ key (Pro+) to provision monitors from code and use a human key (auto-created on first ping) instead of a dashboard slug:

import os, zuvio_atlas as atlas
atlas.configure(api_key=os.environ["ZUVIO_ATLAS_KEY"])

atlas.put("nightly-backup", cron_expression="0 3 * * *")   # idempotent

@atlas.job("nightly-backup")
def run_backup():
    ...

Without an api key, the identifier is a per-monitor slug on the unauthenticated capability URL.

Heartbeats

atlas.heartbeat("important-heartbeat")                       # alive
atlas.heartbeat("important-heartbeat", state="fail", message="disk full")

With an api key the id is a human key (auto-created on first ping); otherwise a slug. atlas.put_heartbeat("key", interval=60) provisions one from code.

Config

atlas.configure(base_url="https://atlas.zuviosystems.com", api_key="pm_live_…", timeout=10.0)
# or set the ZUVIO_ATLAS_URL / ZUVIO_ATLAS_KEY environment variables

Pings are best-effort — a monitoring outage never raises into your job, and your function's return value / exception is always preserved.