77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
|
"""SQLite-backed tracking of active + historical streams.
|
||
|
|
|
||
|
|
Backed by a file on the shared `appdata` PV (mounted at STATE_DB_PATH), so
|
||
|
|
state survives pod restarts/redeploys. "Active" streams are just history
|
||
|
|
rows with end IS NULL — no separate table needed, which also gives free
|
||
|
|
crash recovery (if the pod dies mid-stream, the row is simply still open).
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import threading
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
|
||
|
|
def _now_iso():
|
||
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
|
|
||
|
|
|
||
|
|
SCHEMA = """
|
||
|
|
CREATE TABLE IF NOT EXISTS streams (
|
||
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
|
|
path TEXT NOT NULL,
|
||
|
|
source_type TEXT,
|
||
|
|
start TEXT NOT NULL,
|
||
|
|
end TEXT
|
||
|
|
);
|
||
|
|
CREATE INDEX IF NOT EXISTS idx_streams_active ON streams (path) WHERE end IS NULL;
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
class StreamState:
|
||
|
|
def __init__(self, db_path, history_limit=50):
|
||
|
|
self._db_path = db_path
|
||
|
|
self._history_limit = history_limit
|
||
|
|
self._lock = threading.Lock()
|
||
|
|
os.makedirs(os.path.dirname(db_path) or ".", exist_ok=True)
|
||
|
|
with self._connect() as conn:
|
||
|
|
conn.executescript(SCHEMA)
|
||
|
|
|
||
|
|
def _connect(self):
|
||
|
|
conn = sqlite3.connect(self._db_path)
|
||
|
|
conn.row_factory = sqlite3.Row
|
||
|
|
return conn
|
||
|
|
|
||
|
|
def on_ready(self, path, source_type=None):
|
||
|
|
with self._lock, self._connect() as conn:
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO streams (path, source_type, start, end) VALUES (?, ?, ?, NULL)",
|
||
|
|
(path, source_type, _now_iso()),
|
||
|
|
)
|
||
|
|
|
||
|
|
def on_not_ready(self, path):
|
||
|
|
with self._lock, self._connect() as conn:
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT id FROM streams WHERE path = ? AND end IS NULL ORDER BY id DESC LIMIT 1",
|
||
|
|
(path,),
|
||
|
|
).fetchone()
|
||
|
|
if row:
|
||
|
|
conn.execute(
|
||
|
|
"UPDATE streams SET end = ? WHERE id = ?",
|
||
|
|
(_now_iso(), row["id"]),
|
||
|
|
)
|
||
|
|
|
||
|
|
def active(self):
|
||
|
|
with self._connect() as conn:
|
||
|
|
rows = conn.execute(
|
||
|
|
"SELECT path, source_type, start FROM streams WHERE end IS NULL ORDER BY id DESC"
|
||
|
|
).fetchall()
|
||
|
|
return [dict(r) for r in rows]
|
||
|
|
|
||
|
|
def history(self):
|
||
|
|
with self._connect() as conn:
|
||
|
|
rows = conn.execute(
|
||
|
|
"SELECT path, source_type, start, end FROM streams ORDER BY id DESC LIMIT ?",
|
||
|
|
(self._history_limit,),
|
||
|
|
).fetchall()
|
||
|
|
return [dict(r) for r in rows]
|