Python

Python

cronwatch-sdk is a port of @cronwatch/sdk, not a new design. It decides missed, failed, stuck, slow and over budget by the same rules, sends the same alert text, and writes the same rows, so a Python process, a Node process and a Ruby process can share one database and the MCP server works against any of them. This page covers plain Python and the API underneath.

pip install cronwatch-sdk        # or: uv add cronwatch-sdk

The import name is cronwatch (the name cronwatch on PyPI belongs to an older, abandoned project). Python 3.11 or newer, with no dependencies: cron expressions are read by a port of croner, the parser the SDK uses; zones come from the standard library’s zoneinfo; the SQLite store uses the standard library’s sqlite3. Everything else is an extra, loaded only from its own module:

On Windows, which has no zone database of its own, install cronwatch-sdk[tzdata].

Create one client

import os
import cronwatch
from cronwatch.alerts import Slack
from cronwatch.stores import SqliteStore

cw = cronwatch.Cronwatch(
    store=SqliteStore("./data/cronwatch.db"),
    alerts=[Slack(os.environ["SLACK_WEBHOOK_URL"])],
    retention="30d",
)

Or configure one for the whole process and reach it anywhere with cronwatch.client():

cronwatch.configure(alerts=[Slack(os.environ["SLACK_WEBHOOK_URL"])])

configure takes the same options and returns the client. Configuring again replaces it and stops the old one’s interval checks. cronwatch.client() before any configure is a client with the defaults (the memory store, alerts printed to the console).

Declare and run a job

nightly = cw.job(
    "nightly-report",
    schedule="0 2 * * *", timezone="UTC", grace="15m", timeout="30m",
    expect="Report written", budget={"cost": 2},
)

with nightly.run() as ctx:
    path = build_report()
    ctx.log("Report written:", path)   # kept with the run, shown in alerts
    ctx.metric("cost", 1.2)            # watched against budgets and baselines

The run is recorded when the block ends. An exception inside it is recorded as the failure and raised again, so your own error handling still works; that includes a KeyboardInterrupt or SystemExit that stops the block, so a run is never left running to be reported stuck later.

A function works the same way, passed to run, which returns what the function returns (a string it returns is the run’s output when nothing was logged), or as a decorator, where every call is a run and cronwatch.current() is its context:

rows = nightly.run(lambda ctx: sync_accounts(ctx))

@nightly.monitor
def build_report() -> None:
    cronwatch.current().log("Report written")

Without keeping a handle, cw.run("nightly-report") declares the job on first use (or again, when given options) and works the same way.

Option names are snake_case (max_duration, failures_before_alert). Durations are strings such as "15m" or "1h30m", milliseconds, or a datetime.timedelta. Anything that leaves the process (store rows, webhook bodies) uses the SDK’s camelCase field names, so every language reads it.

The context has name, run_id, started_at, log(*parts), metric(name, value), metrics(dict) and signal, which aborts once the job’s timeout has passed. Nothing is interrupted: a loop that can stop early checks ctx.aborted(), or calls ctx.signal.throw_if_aborted() to raise.

Async jobs

An async def job is wrapped the same way, with async with, an async decorator, or await:

async with nightly.run() as ctx:
    await build_report_async()
    ctx.log("Report written")

@nightly.monitor
async def refresh_cache() -> None: ...

await nightly.run(refresh_cache_async)

cronwatch.aio.AsyncCronwatch(client) (or AsyncCronwatch(store=..., alerts=...)) has the client’s methods as coroutines: await acw.check(), await acw.runs("nightly-report"), await acw.silence("nightly-report", "2h"), and start and resume handles whose finish is awaited. The store, the alert channels and triage run in a worker thread, so a slow database or a hung webhook never blocks the event loop. It needs asyncio (uvicorn, FastAPI, aiohttp); trio is not supported.

Run the check

A long-running process (a web server, a worker) checks in a daemon thread:

cw.start()          # every minute; cw.start("5m") to change it

Calling start again while it runs does nothing; a different interval is reported to on_error and ignored, so call stop() first to change it. A forked child (a Gunicorn or uWSGI worker) gets a fresh client state and no check in flight; call start in the worker, not in the parent before it forks.

Run one checker per store: one process with start, or one scheduled check, not one per process. Two checkers on one database can each send the same alert.

A script run from crontab exits when it is done, so nothing inside it notices the run that never happened. Add a second crontab line that checks:

# check.py
from jobs import cw   # declares every job, so a job that never ran is known

result = cw.check()
print(f"{len(result.jobs)} jobs, {len(result.alerts)} alerts")
cw.close()

0 3 * * *    cd /srv/app && .venv/bin/python backup.py
*/5 * * * *  cd /srv/app && .venv/bin/python check.py

check() returns a result with checked_at, jobs, alerts and pruned. Calls at the same time share one check.

The dashboard

cw.routes() is the dashboard and JSON API, the same pages and endpoints as the TypeScript routes: the board’s counts by health, a timeline of the last day with a lane per job, the table of every job, and for each job its last seven days, runs and definition, all drawn on the server with no script. It is a WSGI app, and .asgi is the same routes as an ASGI app, so it mounts in any Python web server. For Django, see Django.

# Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware
app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {"/cronwatch": cw.routes()})

# FastAPI or Starlette
from starlette.routing import Mount
app.router.routes.append(Mount("/cronwatch", app=cw.routes().asgi))

# Its own WSGI server (gunicorn, uWSGI), at the root
application = cw.routes()

cw.routes(token=..., base_path=..., origin=..., trust_proxy=False):

• token: leave it out to read CRONWATCH_TOKEN; an empty string counts as unset. Without a token, while the environment is development or test, the routes make a token of their own and print a sign-in link to standard output on the first request; anywhere else they answer 503. None opts out and serves them open, for a mount behind your own auth. The environment is the first of CRONWATCH_ENV, APP_ENV and ENVIRONMENT that is set (where the SDK reads NODE_ENV).
• base_path: where it is mounted. It defaults to the mount point the server reports (SCRIPT_NAME, or ASGI’s root_path), so the mounts above need nothing more.
• origin: the public origin, such as "https://app.example.com", to pin it whatever a request says. It then replaces the request’s own for the cross-site check on writes, the cookie’s Secure flag, redirects and the sign-in line.
• trust_proxy: take the origin from the first X-Forwarded-Proto and X-Forwarded-Host, as the SDK’s option does. Off by default.

The token rules, cookie, cross-site rule and every endpoint are the SDK’s; see Dashboard and API. /api/check also accepts the client’s cron_secret as a bearer, so an outside cron can run the check over HTTP. The dashboard is installable as a web app, with its manifest, icons and service worker under the mount point; see Install it as an app. The ASGI app runs each request in a worker thread, so it needs an asyncio server (uvicorn, Hypercorn, Daphne).

Jobs a URL starts

Some platforms run scheduled work by calling a URL: a hosting provider’s cron, a serverless function on a schedule, an outside cron service. job.handler(fn) is that endpoint. It runs fn(ctx, request) as a recorded run for each request carrying Authorization: Bearer <secret>, and answers with JSON saying how the run went:

nightly = cw.job("nightly-report", schedule="0 2 * * *", timezone="UTC")
hook = nightly.handler(lambda ctx, request: build_report())

# Flask
app.add_url_rule("/cron/nightly-report", view_func=hook.flask, methods=["GET", "POST"])

# FastAPI or Starlette
app.add_route("/cron/nightly-report", hook.starlette, methods=["GET", "POST"])

# AWS Lambda behind API Gateway or a function URL, on a schedule
lambda_handler = hook.aws_lambda

.django is a Django view (see Django), .aws_lambda takes Lambda’s (event, context) and returns the response dict API Gateway and function URLs expect, and .wsgi and .asgi are apps of their own. Called directly with a request, the handler answers in that request’s own kind. An async def function gives an async handler.

The secret is handler(fn, secret=...), else the client’s cron_secret, which reads CRON_SECRET by default; it is compared in constant time. A wrong or missing bearer is answered 401 and runs nothing. With no secret at all, outside development, the handler answers 503 and reports it once to on_error, rather than let anyone on the internet run the job; secret=None opts out on purpose, for an endpoint your platform already protects. A run is answered 200 or 500 with {"ok", "job", "run", "status", "durationMs"}, and a function that returns a response of its own is answered with it.

A response with a status of 400 or more fails the run, recorded as HTTP <status> and its reason, whether a handler, run() or finish(result=...) got it. Django, Flask, Starlette and FastAPI, requests, httpx and http.client responses are all recognised, so a job that calls an API and returns its answer fails when the API does.

APScheduler

from apscheduler.schedulers.background import BackgroundScheduler
from cronwatch.apscheduler import watch

scheduler = BackgroundScheduler(timezone="UTC")
scheduler.add_job(nightly_report, "cron", hour=2, id="nightly-report")
watch(scheduler, client=cw)
scheduler.start()

pip install "cronwatch-sdk[apscheduler]". watch adds a listener to an APScheduler 3 scheduler (background, blocking, asyncio or any other) and needs no change to your jobs. Every job becomes a CronWatch job named after its id, with its trigger as the schedule: a cron trigger becomes the same cron expression, checked against APScheduler’s own fire times; an interval trigger is every <interval>; a date trigger runs once. A job added or rescheduled later is followed, and a removed one stops being expected.

Each run is recorded from APScheduler’s own events: the return value is the output when it is a string (and checked by expect), an exception fails it, and a run APScheduler skipped because it could not start within misfire_grace_time is failed with that reason. The events are recorded in a thread of the listener’s own, so neither the scheduler nor an asyncio loop waits on the store. jobs= and exclude= pick the jobs, and other options (grace, timeout, failures_before_alert) apply to each one it declares. cronwatch.current() is not available inside an APScheduler job, since the listener only sees events; log from the job with Python’s logging instead, or wrap its body in job.run() yourself.

APScheduler 4 is still a pre-release that replaced the events and triggers this is built on, so it is not supported yet.

Stores

cronwatch.stores.MemoryStore() is the default. Nothing survives a restart, so a miss cannot be noticed across one, and each process has its own.

cronwatch.stores.SqliteStore(path="./data/cronwatch.db", connection=None, prefix="cronwatch_") keeps everything in one file, in WAL mode. The directory is made if it is missing, and the file and its -wal and -shm files are created private (mode 0600). ":memory:" works too, and connection= takes an open sqlite3 connection of your own, in autocommit mode (isolation_level=None). It creates its tables on first use.

cronwatch.stores.postgres.PostgresStore(conninfo=None, pool=None, prefix="cronwatch_") needs cronwatch-sdk[postgres]. With no connection string it reads DATABASE_URL. It writes through an autocommit connection of its own, so a run recorded inside your transaction is recorded when it happens and stays recorded if that transaction rolls back. It reconnects when the connection breaks, and a forked child opens its own rather than use the parent’s. pool= takes a psycopg_pool.ConnectionPool of yours instead (anything whose connection() is a context manager giving a psycopg connection); close() leaves a pool you passed open. It creates its tables on first use, under an advisory lock, so many processes can start at once.

prefix names the tables (cronwatch_jobs, cronwatch_runs, cronwatch_state): lowercase letters, digits and underscores.

A store of your own is any object with the methods the memory store has: upsert_job, get_job, list_jobs, delete_job, insert_run, update_run, update_run_if, get_run, list_runs, last_run, running_runs, get_state, set_state, compare_and_set_state, prune, and optionally init and close. They mean what the TypeScript interface says, with epoch milliseconds for every time. update_run_if and compare_and_set_state are what keep two processes on one store from judging a run twice or losing each other’s updates; see two processes, one store.

Alerts

import os
from cronwatch import Console, Custom
from cronwatch.alerts import Slack, Discord, Webhook

Slack(os.environ["SLACK_WEBHOOK_URL"], link=lambda a: f"https://app.example.com/cronwatch/jobs/{a.job}")
Discord(os.environ["DISCORD_WEBHOOK_URL"])
Webhook("https://hooks.example.com/cronwatch", secret=os.environ.get("CRONWATCH_WEBHOOK_SECRET"))
Console()

def page(alert):
    if alert.type == "recovered":
        return
    pagerduty.trigger(summary=alert.title, details=alert.message)

Custom("pagerduty", page)

Every alert goes to every channel at once; a channel that raises, or takes longer than 15 seconds, goes to on_error and never holds up the others.

A channel is any object with name and send(alert, context) that raises when the alert went nowhere; a plain function works too, and one that takes only the alert is called with the alert alone. context.on_error(error) reports a problem that did not stop the alert going out (one of several recipients refusing it, say) to the client’s on_error, as "alert channel <name>".

Email, SMS and error trackers

The SDK’s provider channels are in cronwatch.alerts too, on the standard library alone (urllib.request, with SES requests signed by SigV4 and no AWS SDK):

from cronwatch.alerts import (
    Resend, Postmark, Sendgrid, Mailgun, Ses, Twilio,
    Sentry, Honeybadger, Datadog, Rollbar, Bugsnag, NewRelic,
)

# Email. Each takes from_, to (one address or a list), subject_prefix and link.
Resend(api_key=os.environ["RESEND_API_KEY"], from_="CronWatch <alerts@example.com>", to="ops@example.com")
Postmark(server_token=os.environ["POSTMARK_SERVER_TOKEN"], from_="alerts@example.com", to="ops@example.com")
Sendgrid(api_key=os.environ["SENDGRID_API_KEY"], from_="alerts@example.com", to="ops@example.com")
Mailgun(api_key=os.environ["MAILGUN_API_KEY"], domain="mg.example.com", region="eu",
        from_="alerts@example.com", to="ops@example.com")
Ses(region="us-east-1", access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
    secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"], from_="alerts@example.com", to="ops@example.com")

# SMS, one message per number, all at once. Recoveries are not texted unless recovered=True.
Twilio(account_sid=os.environ["TWILIO_ACCOUNT_SID"], auth_token=os.environ["TWILIO_AUTH_TOKEN"],
       from_="+15005550006", to=["+15551110000"])

# Error trackers: one issue per job and alert type.
Sentry(dsn=os.environ["SENTRY_DSN"])
Honeybadger(api_key=os.environ["HONEYBADGER_API_KEY"])
Datadog(api_key=os.environ["DD_API_KEY"], site="datadoghq.eu", tags=["env:prod"])
Rollbar(access_token=os.environ["ROLLBAR_ACCESS_TOKEN"])
Bugsnag(api_key=os.environ["BUGSNAG_API_KEY"])
NewRelic(account_id=os.environ["NEW_RELIC_ACCOUNT_ID"], api_key=os.environ["NEW_RELIC_LICENSE_KEY"])

from is a Python keyword, so the sender is from_. The other options are the SDK’s in snake_case: subject_prefix, message_stream (Postmark), region ("eu" for SendGrid, Mailgun and New Relic; the AWS region for SES), session_token and configuration_set_name (SES), api_key_sid, api_key_secret, messaging_service_sid and segments (Twilio), environment and release (Sentry), endpoint (Honeybadger, Bugsnag), host (Datadog), release_stage (Bugsnag), event_type (New Relic), and recovered and link wherever the SDK has them. A missing key, address or account raises ValueError when the channel is made.

Each sends exactly the request the SDK’s does: the same URL, headers and body, byte for byte (the package’s tests replay the SDK’s recorded requests), with the same idempotency key, event id or UUID for one alert, so a provider that deduplicates drops a resend whichever language sent it. Each request gives up 10 seconds after it starts. A refused request raises "<Provider> <origin> answered <status>: <start of the body>", never the URL’s path, with the channel’s keys cut out of the quoted body. No channel follows a redirect, so credentials never reach another address. Alerts describes what each one sends. One difference from Node: urllib honours the HTTP_PROXY and HTTPS_PROXY environment variables, which Node’s fetch ignores by default.

A webhook signs its body with X-CronWatch-Signature: sha256=<hex>. Verifying it in Python:

import hashlib, hmac

expected = "sha256=" + hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
ok = hmac.compare_digest(expected, request.headers.get("X-CronWatch-Signature", ""))

Processes that cannot send

A job can run somewhere that cannot reach Slack or a mail relay: a sandboxed worker, a script without the app’s secrets. Give that process deliver="check":

recorder = cronwatch.Cronwatch(store=PostgresStore(), deliver="check")

It still records and evaluates every run, but queues each alert in the store instead of sending it. The next check in a process that sends normally delivers it, with triage if that process has it. Both processes must use the same store. See processes that cannot send.

pg_cron

pg_cron runs jobs inside Postgres, where nothing can wrap them. PgCron reads what pg_cron records instead: on every check it reads cron.job, declares each job with its schedule, and copies new rows of cron.job_run_details in as runs, so a job that stops running is missed, a failed run alerts and a run that never ends is stuck.

from cronwatch.sources.pgcron import PgCron
from cronwatch.stores.postgres import PostgresStore

cw = cronwatch.Cronwatch(
    store=PostgresStore(),
    sources=[PgCron(os.environ["DATABASE_URL"], prefix="db:")],
)
cw.start()

The first argument is a connection string, a psycopg connection or pool, or anything with query(sql, params) that returns rows as dicts. On a connection that is not in autocommit mode it never ends a transaction of yours. The options (jobs, prefix, job_name, options, timezone) and the rules for renamed jobs, runs cut off by a restart and history seen for the first time are the SDK’s; see Supabase and pg_cron.

Redaction

Before a run’s output and error are stored, shown or sent anywhere, redact rewrites them. The default, cronwatch.redact_secrets, blanks values that look like secrets (secret-named pairs, credentials in URLs, authorization headers, private keys, JWTs, webhook URLs, and AWS, GitHub, Slack, Stripe, Google and API key formats), exactly what the SDK’s default blanks. An expect rule is checked before redaction, so it still sees what was logged.

cronwatch.Cronwatch(redact=False)                                            # keep output as logged
cronwatch.Cronwatch(redact=lambda text: re.sub(r"\d{16}", "[card]", cronwatch.redact_secrets(text)))

A redact that raises, or returns something other than a str, is reported to on_error (as "redact") and the default is used for that text.

Triage

pip install "cronwatch-sdk[anthropic]"

from cronwatch.triage.anthropic import AnthropicTriage

cw = cronwatch.Cronwatch(triage=AnthropicTriage(context="A Flask app on Postgres, jobs run from crontab."))

It sends the same request as the SDK’s, runs only when an alert is sent (never per run, never for a recovery), and once per alert. The client waits 25 seconds for it, then sends the alert without a diagnosis and reports the timeout to on_error. What is sent is in AI triage.

A triage of your own is any function that takes the context (alert, recent_runs, signal) and returns a string or None.

API

cronwatch.Cronwatch(...) and cronwatch.configure(...):

cw.job(name, **options) takes schedule (five or six field cron, a nickname such as "@hourly", or "every 5m"), timezone (IANA; the process’s zone by default), grace ("10m"), timeout ("1h"), max_duration, budget ({"metric": ceiling}), expect (a string the output must contain, a compiled re pattern it must match, or a function), failures_before_alert (1), description and tags, with the rules in the API reference. A name is 1 to 120 letters, digits, ., _, : or -. Bad options raise ValueError when the job is declared.

The client:

Runs that span calls

A run is normally one call. Work that starts in one place and ends in another (a job that hands work to a queue, a webhook that reports completion later) can be one run too: start records it as running and returns a run handle, and finish on that handle, or on one from resume(run_id) in another process, ends it.

sync = cw.job("partner-sync", schedule="0 * * * *", timeout="2h")

run = sync.start(id=batch_id)       # records a running run
# later, perhaps in another process
run = sync.resume(batch_id)         # or cw.resume_run("partner-sync", batch_id)
run.log("imported", count, "rows")
run.finish()                        # or run.fail(error), or run.finish(result="text")

start(trigger=None, id=None) takes your own stable id, 1 to 200 characters: a start with an id already recorded for this job records nothing and returns a handle on that run. A store that fails is reported to on_error, never raised. The handle has log, metric, metrics, flush() (append what is logged so far to the stored run), finish() and fail(error), and active(), false once it is finished. A run is judged once however many times it is finished: a second finish, or one on a run another process finished, records nothing and is reported to on_error. A run that is never finished is marked stuck by the first check after the job’s timeout, so set timeout to cover the whole span. The Ruby page has the full rules, which are the same.

Sharing a database with Node and Ruby

The SQLite and Postgres stores write the same three tables as @cronwatch/sdk/sqlite and @cronwatch/sdk/postgres and the Ruby gem’s store: the same names, columns and indexes, epoch milliseconds in the time columns, and the same JSON in the JSON columns. The package’s tests share a SQLite file with the built SDK and check that each side reads what the other wrote. Create the tables from any side; the others find them and leave them alone. Use the same prefix everywhere.

Each process alerts on the jobs it runs, and any side’s check sees every job in the store. One dashboard shows them all, and one MCP server reads it. Give each job a name only one side uses, and run one checker for the store.

The cron reader matches croner with two exceptions, both for schedules that never make sense: a date no month has (0 0 30 2 *) is a schedule that never fires, where croner gives up; and a one-time date in place of a cron expression (2026-12-01T00:00:00) is refused.

Kept in step

The TypeScript SDK is the source of truth. Its build generates cases (duration parsing, schedules across daylight saving, sequences of runs and checks with the alerts and state they must produce, alert titles and messages, each channel’s requests, stats and health) into conformance/ in the repository, and the Python package’s tests replay every one, as the Ruby gem’s do. A change of behaviour lands in TypeScript first, the cases are regenerated, and the port is fixed until they pass. Where they disagree, the port is wrong: open an issue.