Worum es geht
Eine Überwachungs-Pipeline liest nicht nur Feeds. Viele selbst gehostete Anwendungen bieten eine HTTP-API, über die sich Daten abfragen lassen: Medienserver, Wiki-Systeme, Fitness-Tracker. Ein Beispiel dafür ist LiftTrace, ein selbst gehosteter Tracker für Krafttraining, der laut öffentlicher Beschreibung in einem einzelnen Docker-Container läuft und unter der AGPL steht. Welche Endpunkte und Funktionen er tatsächlich bietet, entnimmst du der Dokumentation des Projekts. Hier geht es um das allgemeine Muster, das für jede solche Anwendung gilt.
Der Nutzen: Neue Einträge in einer Anwendung lösen eine Meldung aus, oder du sammelst Daten für eine spätere Auswertung.
Das Muster: Polling per cron
Viele Anwendungen bieten keine Webhooks. Dann fragt ein kleiner Poller in festen Abständen nach, ob es etwas Neues gibt. Das ist einfach, robust und braucht keine Änderung an der Anwendung.
Der Ablauf bei jedem Lauf:
- API-Schlüssel und Adresse aus der Umgebung lesen.
- Den Zeitstempel des letzten erfolgreichen Abrufs aus SQLite holen.
- Nur Einträge seit diesem Zeitpunkt abfragen (inkrementell).
- Neue Einträge speichern und eine Meldung auslösen.
- Den Zeitstempel erst nach erfolgreichem Speichern fortschreiben.
Schlüssel und Adresse aus der Umgebung
API-Schlüssel gehören nicht in den Code. Lege sie in einer Umgebungsdatei ab, die nur der ausführende Benutzer lesen darf, und lade sie über die Unit oder das Startskript.
Ein Poller als Beispiel
Die Adresse und der Endpunkt im Beispiel sind Platzhalter, die du an die jeweilige Anwendung anpasst.
import logging
import os
import sqlite3
from datetime import datetime, timezone
import requests
log = logging.getLogger("poller")
BASE_URL = os.environ["SOURCE_URL"]
API_KEY = os.environ["SOURCE_API_KEY"]
def last_seen(db: sqlite3.Connection) -> str:
row = db.execute("SELECT value FROM state WHERE key='since'").fetchone()
return row[0] if row else "1970-01-01T00:00:00+00:00"
def fetch_since(since: str) -> list[dict]:
r = requests.get(f"{BASE_URL}/api/entries",
params={"since": since},
headers={"Authorization": f"Bearer {API_KEY}"},
timeout=15)
r.raise_for_status()
return r.json()
def run(db: sqlite3.Connection, notify) -> None:
since = last_seen(db)
started = datetime.now(timezone.utc).isoformat()
try:
entries = fetch_since(since)
except requests.RequestException as exc:
log.warning("Abruf fehlgeschlagen: %s", exc)
return
for entry in entries:
db.execute("INSERT OR IGNORE INTO entries(id, payload) VALUES (?, ?)",
(entry["id"], str(entry)))
db.execute("INSERT OR REPLACE INTO state(key, value) VALUES ('since', ?)",
(started,))
db.commit()
if entries:
notify(f"{len(entries)} neue Einträge")
Der Zeitstempel wird vor dem Abruf genommen und erst nach dem Speichern gesichert. So gehen bei einem Absturz keine Einträge verloren, im schlimmsten Fall werden welche doppelt gelesen. INSERT OR IGNORE fängt das ab.
Einbindung per cron oder Timer
Ein Aufruf alle 30 Minuten genügt für die meisten Zwecke. Häufigeres Abfragen belastet die Anwendung unnötig. Protokolliere Fehler in das Journal, damit ein Ausfall der Quelle auffällt.
Was du beachten solltest
- Lizenz: Bei Anwendungen unter AGPL gelten Bedingungen, wenn du sie öffentlich anbietest oder veränderst. Für die private Nutzung im eigenen Netz und das reine Abfragen der API ändert sich dafür in der Regel nichts, lies aber die Lizenz des jeweiligen Projekts.
- Daten: Trainings- oder andere persönliche Daten sind sensibel. Behandle die Datenbank der Pipeline entsprechend.
- Stabilität: Junge Projekte ändern ihre API. Kapsle den Zugriff in einer Funktion, damit du Anpassungen an einer Stelle machst.
- Volumen: Frage nur nach dem Nötigen und begrenze die Zahl der Einträge pro Abruf.
Wann sich das lohnt (und wann nicht)
Das Muster lohnt sich, wenn du Daten einer eigenen Anwendung mit deiner Pipeline verbinden willst, etwa für Erinnerungen, Zusammenfassungen oder Auswertungen. Der Poller ist klein und lässt sich für jede Quelle mit HTTP-API wiederverwenden.
Wenn die Anwendung Webhooks anbietet, sind diese oft die bessere Wahl, weil sie ohne Abfragen auskommen. Und wenn du die Daten nur ansehen willst, braucht es keine Pipeline, die Oberfläche der Anwendung reicht.