What this is about
A monitoring pipeline does not only read feeds. Many self-hosted applications offer an HTTP API through which data can be queried: media servers, wiki systems, fitness trackers. One example is LiftTrace, a self-hosted tracker for strength training that, according to its public description, runs in a single Docker container and is licensed under the AGPL. Which endpoints and features it actually offers is found in the project's documentation. This article is about the general pattern, which holds for any such application.
The benefit: new entries in an application trigger a notification, or you collect data for a later evaluation.
The pattern: polling via cron
Many applications offer no webhooks. Then a small poller asks at fixed intervals whether there is anything new. This is simple, robust and needs no change to the application.
The flow on every run:
- Read the API key and address from the environment.
- Fetch the timestamp of the last successful retrieval from SQLite.
- Query only entries since that point (incrementally).
- Store new entries and trigger a notification.
- Advance the timestamp only after storing has succeeded.
Key and address from the environment
API keys do not belong in the code. Put them in an environment file that only the running user may read, and load it through the unit or the start script.
A poller as an example
The address and the endpoint in the example are placeholders that you adapt to the respective application.
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")
The timestamp is taken before the retrieval and secured only after storing. That way no entries are lost on a crash, and in the worst case some are read twice. INSERT OR IGNORE catches that.
Running it via cron or a timer
A call every 30 minutes is enough for most purposes. Querying more often puts unnecessary load on the application. Write errors to the journal so that an outage of the source gets noticed.
What to keep in mind
- License: for applications under the AGPL, conditions apply if you offer them publicly or modify them. For private use in your own network and merely querying the API this usually changes nothing, but read the license of the respective project.
- Data: training data and other personal data are sensitive. Treat the pipeline's database accordingly.
- Stability: young projects change their API. Encapsulate access in one function so that you make adjustments in one place.
- Volume: ask only for what is needed and limit the number of entries per call.
When it is worth it (and when not)
The pattern is worth it when you want to connect data from one of your own applications with your pipeline, for example for reminders, summaries or evaluations. The poller is small and can be reused for any source with an HTTP API.
If the application offers webhooks, they are often the better choice because they need no polling. And if you only want to look at the data, you do not need a pipeline, the application's interface is enough.