import json import sqlite3 from contextlib import contextmanager from .config import DB_PATH # Ein Eintrag pro Termin, Todo oder Notiz. Gelöschte bleiben als Tombstone erhalten, # sonst würde ein anderes Gerät sie wieder hochladen. # updated: Zeitstempel des Geräts, entscheidet bei Konflikten (neuester gewinnt). # seq: Server-Zähler, damit Clients "alles seit meinem letzten Stand" abholen können. SCHEMA = """ CREATE TABLE IF NOT EXISTS items ( collection TEXT NOT NULL, id TEXT NOT NULL, updated TEXT NOT NULL, deleted INTEGER NOT NULL DEFAULT 0, data TEXT NOT NULL, seq INTEGER NOT NULL, PRIMARY KEY (collection, id) ); CREATE INDEX IF NOT EXISTS items_seq ON items(seq); """ # Große Datei-Uploads halten die Datenbank sekundenlang. WAL lässt Leser derweil durch, # Schreiber warten bis zu 60 s statt mit "database is locked" abzubrechen. @contextmanager def connect(): DB_PATH.parent.mkdir(parents=True, exist_ok=True) conn = sqlite3.connect(DB_PATH, timeout=60) conn.row_factory = sqlite3.Row try: yield conn conn.commit() finally: conn.close() def init(): with connect() as conn: conn.executescript(SCHEMA) conn.execute("PRAGMA journal_mode=WAL") def cursor(conn): return conn.execute("SELECT COALESCE(MAX(seq), 0) FROM items").fetchone()[0] def merge(conn, collection, item): """Übernimmt den Eintrag, wenn er neuer ist als der gespeicherte. Gibt (True, None) zurück, wenn er übernommen wurde, sonst (False, gespeicherter Eintrag), damit der Client den gültigen Stand bekommt. """ row = conn.execute( "SELECT updated, data FROM items WHERE collection = ? AND id = ?", (collection, item["id"]) ).fetchone() if row and row["updated"] >= item["updated"]: return False, json.loads(row["data"]) conn.execute( "INSERT OR REPLACE INTO items (collection, id, updated, deleted, data, seq) VALUES (?, ?, ?, ?, ?, ?)", ( collection, item["id"], item["updated"], 1 if item.get("deleted") else 0, json.dumps(item, ensure_ascii=False), cursor(conn) + 1, ), ) return True, None def since(conn, collection, seq): rows = conn.execute( "SELECT data FROM items WHERE collection = ? AND seq > ? ORDER BY seq", (collection, seq) ) return [json.loads(r["data"]) for r in rows] def since_paged(conn, collection, seq, max_bytes): """Wie since, aber höchstens max_bytes Daten. Gibt (Einträge, letzter seq, mehr da) zurück.""" rows = conn.execute( "SELECT data, seq FROM items WHERE collection = ? AND seq > ? ORDER BY seq", (collection, seq) ) items, size, last = [], 0, seq for r in rows: if items and size + len(r["data"]) > max_bytes: return items, last, True items.append(json.loads(r["data"])) size += len(r["data"]) last = r["seq"] return items, last, False def alive(conn, collection): rows = conn.execute( "SELECT data FROM items WHERE collection = ? AND deleted = 0 ORDER BY id", (collection,) ) return [json.loads(r["data"]) for r in rows]