Documentation menu

Python

Two setups cover most Python services:

You create monitors in the dashboard under Monitors → New monitor. The code below only sends pings or answers checks, so it needs no API key.

Heartbeat: report a job

Create a heartbeat monitor, copy its ping URL and set it as an environment variable. Keep it out of source control: anyone with the URL can send pings.

ASNIHER_HEARTBEAT_URL=https://app.asniher.com/api/heartbeat/<token>

Save this helper as asniher.py. It uses only the standard library and works on Python 3.8 and later. It pings /start before the job, the success URL after it, and /fail with the traceback if the job raises. A ping that cannot reach Asniher prints a warning and never fails your job.

import os
import sys
import traceback
import urllib.error
import urllib.request
from contextlib import contextmanager

URL = os.environ.get("ASNIHER_HEARTBEAT_URL")


def ping(suffix="", output=None):
    if not URL:
        return
    data = output.encode()[-10_000:] if output else b""
    request = urllib.request.Request(URL + suffix, data=data, method="POST")
    for attempt in range(3):
        try:
            urllib.request.urlopen(request, timeout=10).close()
            return
        except urllib.error.HTTPError as exc:
            print(f"Asniher ping returned {exc.code}", file=sys.stderr)
            return
        except OSError as exc:
            if attempt == 2:
                print(f"Asniher ping failed: {exc}", file=sys.stderr)


@contextmanager
def monitored():
    ping("/start")
    try:
        yield
    except Exception:
        ping("/fail", traceback.format_exc())
        raise
    ping()

Wrap the job:

from asniher import monitored

with monitored():
    export_orders()

Run it from cron and copy the same cron expression into the monitor's schedule:

30 2 * * * cd /srv/app && ASNIHER_HEARTBEAT_URL=https://app.asniher.com/api/heartbeat/<token> python3 jobs/export_orders.py

The traceback is sent as job output, so it shows up in the ping log and the alert. Avoid putting secrets in exception messages.

Report a warning

If a run finished but needs attention, send /warn with a short explanation. The monitor stays up and you get a warning alert.

import shutil
from asniher import ping

free_gb = shutil.disk_usage("/home").free // 1024**3
if free_gb < 10:
    ping("/warn", f"/home has {free_gb} GB free")
else:
    ping()

Django management commands and Celery

The context manager works anywhere. Use it inside a management command's handle() that cron runs, or inside a Celery task:

from celery import shared_task
from asniher import monitored


@shared_task
def nightly_report():
    with monitored():
        build_report()

A Celery task that beat fails to schedule never runs and never pings, so the monitor alerts. That covers a stopped beat process as well as a failing task.

HTTP monitor: health endpoint

An HTTP monitor needs a public URL that returns 2xx when the service works and something else when it does not. Check the dependencies the service cannot work without, and return 503 if one is down.

FastAPI

import os

from fastapi import FastAPI, Response
from sqlalchemy import create_engine, text

app = FastAPI()
engine = create_engine(os.environ["DATABASE_URL"], pool_pre_ping=True)


@app.get("/healthz")
def healthz(response: Response):
    response.headers["cache-control"] = "no-store"
    try:
        with engine.connect() as conn:
            conn.execute(text("select 1"))
    except Exception:
        response.status_code = 503
        return {"status": "unavailable"}
    return {"status": "ok"}

Django

# health/views.py
from django.db import connection
from django.http import JsonResponse
from django.views.decorators.cache import never_cache


@never_cache
def healthz(request):
    try:
        with connection.cursor() as cursor:
            cursor.execute("select 1")
    except Exception:
        return JsonResponse({"status": "unavailable"}, status=503)
    return JsonResponse({"status": "ok"})
# urls.py
from django.urls import path
from health.views import healthz

urlpatterns = [
    path("healthz", healthz),
]

Then create an HTTP monitor:

Setting Value
URL https://api.example.com/healthz
Method GET
Expected status 200
Interval Your plan's minimum

Keep the endpoint fast and free of secrets. Asniher identifies itself as AsniherBot/1.0; allow that user agent if bot protection sits in front of the endpoint. Add the host to ALLOWED_HOSTS in Django.

Check that it works

python3 -c "import asniher; asniher.ping()"

The ping appears in the monitor's ping log within seconds. Nothing in the log means the variable is not set or the token is wrong; curl -fsS -X POST "$ASNIHER_HEARTBEAT_URL" prints the error.

Questions about this page? Email support@asniher.com.