← API

Cookbook: watch the source catalog

A feed is only as good as the collection behind it. Every source in the catalog reports when it last synced successfully — this check polls that field and flags any enabled source that has been quiet longer than your threshold.

The pattern

• One request. https://pubsec.pro/api/v1/sources returns the whole catalog, no pagination.
• Skip enabled: false. A source switched off is a decision, not an incident.
• Compare last_success against a staleness threshold — 48 hours below; pick your own.
• Alert by exit code. Print the stale sources and exit nonzero; cron emails you the output. No mail code required.

About that timestamp

last_success looks like 2026-10-05 17:26:20.101271+00 — a space between date and time, six-digit microseconds, and a two-digit UTC offset. Python's datetime.fromisoformat rejects the short offset on 3.10 and older, so the script splits the string and rebuilds a full ISO timestamp before parsing. It runs unchanged on Python 3.8+.

The script

Save as source_monitor.py and run it hourly: 0 * * * * python3 /path/to/source_monitor.py.

#!/usr/bin/env python3
"""Source-staleness monitor: check the catalog, flag quiet sources, exit nonzero.

Cron entry (top of every hour):
    0 * * * * python3 /path/to/source_monitor.py
"""
import json
import urllib.request
from datetime import datetime, timedelta, timezone

STALE_AFTER = timedelta(hours=48)   # pick your own threshold
CATALOG = "https://pubsec.pro/api/v1/sources"


def parse_last_success(ts):
    # "2026-10-05 17:26:20.101271+00" -> aware UTC datetime.
    # The two-digit offset breaks fromisoformat() on Python <= 3.10,
    # so split the string and rebuild a full ISO timestamp.
    date_part, clock, offset = ts.split(" ")
    return datetime.fromisoformat(f"{date_part}T{clock}{offset}:00").astimezone(timezone.utc)


def main():
    with urllib.request.urlopen(CATALOG) as response:
        catalog = json.load(response)

    now = datetime.now(timezone.utc)
    problems = []
    for source in catalog["data"]:
        if not source["enabled"]:
            continue                    # switched off on purpose: not an alert
        if not source["last_success"]:
            problems.append((source["source_id"], "never succeeded"))
            continue
        age = now - parse_last_success(source["last_success"])
        if age > STALE_AFTER:
            hours = age.total_seconds() / 3600
            problems.append((source["source_id"], f"last success {hours:.0f} h ago"))

    if not problems:
        print(f"OK - {catalog['meta']['count']} sources in the catalog, "
              f"no enabled source stale (threshold {STALE_AFTER})")
        return

    print(f"STALE SOURCES (threshold {STALE_AFTER}):")
    for source_id, why in problems:
        print(f"  {source_id}: {why}")
    raise SystemExit(1)                 # nonzero exit -> cron emails the output


if __name__ == "__main__":
    main()

Notes

• Stateless. One API request per run, nothing stored — run it as often as you like.
• Never-succeeded is its own alert. A source with last_success: null and enabled: true gets flagged, not skipped.
• Prefer real email? Swap the print loop for the SMTP block from the matched-tenders digest; the shape is identical.

Back to the API reference.