Overview

A hook is any callable taking (job_name, context). Both sync and async hooks work. There are three kinds:

Hook	Registered with	Runs
Before run	`add_before_run_hook()`	Immediately before the job body
After run	`add_after_run_hook()`	After the job returns successfully
On error	`add_on_error_hook()`	When the job raises, including on timeout

Registering hooks

Hooks registered on the scheduler apply to every job:

def log_start(job_name: str, context: dict):
    print(f"starting {job_name} (scheduled {context['scheduled_time']})")


async def notify_success(job_name: str, context: dict):
    await send_slack(f"{job_name} finished in {context['duration']:.2f}s")


async def handle_error(job_name: str, context: dict):
    await page_on_call(job_name, context["error"])


crons.add_before_run_hook(log_start)
crons.add_after_run_hook(notify_success)
crons.add_on_error_hook(handle_error)

Pass job_name to scope a hook to one job, or attach it to the job object directly:

# Only for the "backup_database" job.
crons.add_before_run_hook(log_start, job_name="backup_database")

# Or attach directly to the job object.
job = crons.get_job("backup_database")
job.add_after_run_hook(notify_success)

Hook methods chain

add_before_run_hook() and friends return the Crons instance, so registrations can be chained.

The execution context

Every hook receives a dictionary describing the run. Which keys are present depends on when the hook fires.

Always present

job_name

str

Name of the job.

After run and on error

start_time

datetime

When the run began.

end_time

datetime

When the run finished.

duration

float

Elapsed seconds.

success

bool

Whether the job completed without raising.

result

Any

The job's return value. After-run hooks only.

error

str

The exception message. Error hooks only.

A failing hook does not fail the job

Exceptions raised inside a hook are logged and swallowed so one bad listener cannot take down the scheduler. Do not rely on a hook to enforce correctness of the job itself.

Built-in hooks

Common cases ship with the package, so you rarely need to write logging or alerting hooks by hand.

from fastapi_crons import (
    alert_on_failure,
    alert_on_long_duration,
    log_job_error,
    log_job_start,
    log_job_success,
    webhook_notification,
)

crons.add_before_run_hook(log_job_start)
crons.add_after_run_hook(log_job_success)
crons.add_on_error_hook(log_job_error)

# Alerting
crons.add_on_error_hook(alert_on_failure)
crons.add_after_run_hook(alert_on_long_duration(threshold_seconds=60))

# POST the context to your own endpoint
crons.add_on_error_hook(webhook_notification("https://example.com/hooks/cron"))

Hook	Kind	Purpose
`log_job_start`	Before	Log that a job started.
`log_job_success`	After	Log a successful run and its duration.
`log_job_error`	Error	Log the failure and traceback.
`alert_on_failure`	Error	Emit an alert on any failure.
`alert_on_long_duration(threshold_seconds)`	After	Factory. Alerts when a run exceeds the threshold.
`webhook_notification(url, include_context=True)`	Any	Factory. POSTs the context as JSON to `url`.

Two of these are factories

alert_on_long_duration and webhook_notification return a hook — call them when registering rather than passing the function itself.

Collecting metrics

metrics_collector is a ready-made instance that counts runs, successes and failures, and tracks durations per job.

from fastapi_crons import metrics_collector

crons.add_before_run_hook(metrics_collector.record_job_start)
crons.add_after_run_hook(metrics_collector.record_job_success)
crons.add_on_error_hook(metrics_collector.record_job_failure)


@app.get("/metrics/crons")
def cron_metrics():
    return metrics_collector.get_metrics()

get_metrics() returns the totals for every job; get_job_metrics(name) narrows it to one.

Metrics are in-process only

The counters live in memory and reset when the process restarts. They are not shared between replicas — export them to Prometheus or your own store if you need durable numbers.