Background
Once you run more than a handful of small static sites, you want one screen that answers "what is the state of everything right now" — analytics sessions, Search Console impressions, what got updated today, what the cloud bill is at.
So I built one. Each run of the collector produces a single snapshot, the server stores it, and history accumulates.
The constraint I set was that the server side gets no dependencies at all. No Flask, nothing installed. Just Python's http.server and sqlite3 handling ingest, storage and HTML delivery. I wanted deployment to be "copy the file, run python3 server.py", whether that is a small Linux box or my laptop.
How it works
Three pieces:
- A collector (Node) runs twice a day at 10:00 and 22:00, gathers analytics, Search Console, per-site updates and cloud cost into one JSON object, and
POSTs it to/api/ingest. - The receiver (Python
http.server) stores that payload as one snapshot in SQLite. - The page reads the latest snapshot and renders bars in plain CSS, with a detail modal per site.
All the gathering logic lives in the collector. The receiver only authenticates, stores and retrieves — which is why it has needed almost no changes as sites were added.
Implementation
Dependency-free HTTP with token auth
Subclass BaseHTTPRequestHandler and implement POST /api/ingest. Writes always require a token in an X-Token header, read from an environment variable or a .ingest_token file sitting next to the script, so nobody can push snapshots from outside.
import json, os, sqlite3
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path != "/api/ingest":
return self.send_error(404)
if self.headers.get("X-Token") != TOKEN: # writes need a token
return self.send_error(403)
n = int(self.headers.get("Content-Length", 0))
payload = json.loads(self.rfile.read(n) or b"{}")
sid = ingest(payload)
self._json({"ok": True, "snapshot_id": sid})
ThreadingHTTPServer rather than the single-threaded one, so an ingest arriving while a page is being rendered does not block.
A successful push looks like this from the collector's side — the returned snapshot_id is the row that was just created, and it increments twice a day:
push ok -> snapshot_id=101 (sites=23 / GA=true / SC=true / GCP=130 JPY / tweets_today=19)
A normalized schema keyed by snapshot id
Dropping the whole POST body into one JSON column is tempting and costs you later: every chart has to parse it again. Instead, one row in snapshots, and every other concern joined to it by snapshot_id — per-site metrics, per-service cloud cost, the day's updates. Time-series aggregation then stays pure SQL.
def init_db():
c = db()
c.executescript("""
CREATE TABLE IF NOT EXISTS snapshots(
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT, ts_jst TEXT, gcp_cost TEXT, meta TEXT);
CREATE TABLE IF NOT EXISTS site_metrics(
snapshot_id INTEGER, site_key TEXT, label TEXT,
ga_sessions_y REAL, ga_pv_y REAL,
sc_clicks REAL, sc_impressions REAL, sc_ctr REAL, sc_position REAL);
CREATE TABLE IF NOT EXISTS gcp_services(
snapshot_id INTEGER, service TEXT, cost REAL);
CREATE INDEX IF NOT EXISTS idx_sm_snap ON site_metrics(snapshot_id);
""")
c.commit(); c.close()
def ingest(p):
c = db(); cur = c.cursor()
cur.execute("INSERT INTO snapshots(ts,ts_jst,gcp_cost,meta) VALUES(?,?,?,?)",
(p.get("ts"), p.get("ts_jst"), p.get("gcp_cost"),
json.dumps(p.get("meta") or {})))
sid = cur.lastrowid
for s in p.get("sites", []):
cur.execute("INSERT INTO site_metrics VALUES(?,?,?,?,?,?,?,?,?)",
(sid, s.get("key"), s.get("label"), s.get("ga_sessions_y"),
s.get("ga_pv_y"), s.get("sc_clicks"), s.get("sc_impressions"),
s.get("sc_ctr"), s.get("sc_position")))
c.commit(); c.close(); return sid
Metrics with a settled shape go in columns; loose extras go in the meta JSON column. That split keeps the schema firm without making every new field a migration.
With that layout, the trend chart is one query — total sessions per snapshot, most recent 60:
SELECT s.id sid, s.ts_jst tsj, COALESCE(SUM(m.ga_sessions_y),0) tot
FROM snapshots s
LEFT JOIN site_metrics m ON m.snapshot_id = s.id
GROUP BY s.id ORDER BY s.id DESC LIMIT 60
Gotchas
Positional inserts break the day you add a column
The INSERT INTO site_metrics VALUES(?,?,?,...) above is the version that eventually bit me. Adding one column to the collector made every push fail with table has N columns but M values supplied. The comment now sitting above that statement:
# Always INSERT into site_metrics with explicit column names.
# With positional `VALUES(?,?,...)`, adding a single column makes every push
# fail with "table has N columns but M values supplied" -- an operational
# incident, and worse than breaking quietly would have been.
SITE_METRIC_COLS = ["snapshot_id", "site_key", "label", "url",
"ga_sessions_y", "ga_users_y", "ga_pv_y", "ga_sessions_7d", "ga_pv_7d",
"sc_clicks", "sc_impressions", "sc_ctr", "sc_position", "added",
"ga_organic_y", "ga_organic_7d", "ga_organic_engaged_7d"]
Tables whose shape is still moving get named columns. Small, stable ones keep the positional form. The rule is not "always name columns" but "name them where the shape is still changing."
Migrations only ever add
When the collector starts sending a new field, an old database lacks the column and everything falls over. At startup, check PRAGMA table_info and ALTER TABLE ... ADD COLUMN anything missing:
def ensure_columns(c, table, cols):
"""Adds only the columns that do not exist yet. Idempotent, and never
rewrites existing rows (new columns become NULL, which is correct:
that push simply did not collect them)."""
have = {r["name"] for r in c.execute(f"PRAGMA table_info({table})").fetchall()}
for name, decl in cols:
if name not in have:
c.execute(f"ALTER TABLE {table} ADD COLUMN {name} {decl}")
Adding a column in SQLite is a metadata-only operation, so history survives untouched. Add columns, never drop them, and any build works against any database.
Do not deduplicate the same day
Two pushes a day means two rows a day. Rather than upserting the second over the first, every push appends. Trend and "change since last run" only exist because the history does. There is no unique constraint on the timestamp and that is deliberate.
Fix the timezone at write time
Store ts as ISO UTC and also write ts_jst ("YYYY/MM/DD HH:MM JST") at insert time, keeping both. If the display layer converts on every render, one of those conversions eventually drifts.
Missing is not zero
Every numeric column distinguishes them, and the renderer does too:
def mval(v, title="欠測（取得できず）。実測ゼロ(0)ではありません"):
"""Missing (None) renders as an em dash; a measured zero renders as 0.
Never silently collapse one into the other."""
NULL means the collector failed to fetch. 0 means it fetched successfully and the number was zero. Merging them turns a broken credential into "traffic disappeared", which is a conclusion you will act on.
The result
The dashboard sits alongside the other tools I have built: https://hashitosystem.com
Wrap-up
http.server plus sqlite3 is genuinely enough for an internal ops dashboard. The division of labour is what makes it hold up:
- Collection lives in the collector. The receiver has no idea what an "impression" is.
- Settled metrics become columns; loose extras go in a
metaJSON column. - Timezone is resolved at write time, not at render time.
- Migrations only add columns, so old databases keep working.
- Append every push, because history is the entire point of snapshots.
The receiver has barely changed as the number of sites grew, and that is the whole return on keeping it dependency-free and logic-free.
This article is about my own side project. It was written with AI assistance.
Top comments (0)