Joseph Doherty
39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
from __future__ import annotations
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any, Iterator
|
|
from sqlite3 import Connection
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
id: int
|
|
branch_id: int
|
|
ts: str
|
|
kind: str
|
|
payload: dict[str, Any]
|
|
superseded_by: int | None
|
|
hidden: bool
|
|
|
|
|
|
def append_event(conn: Connection, *, kind: str, payload: dict[str, Any], branch_id: int = 1) -> int:
|
|
cur = conn.execute(
|
|
"INSERT INTO event_log (branch_id, kind, payload_json) VALUES (?, ?, ?)",
|
|
(branch_id, kind, json.dumps(payload)),
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
def read_events(conn: Connection, branch_id: int = 1, after_id: int = 0) -> Iterator[Event]:
|
|
cur = conn.execute(
|
|
"SELECT id, branch_id, ts, kind, payload_json, superseded_by, hidden "
|
|
"FROM event_log WHERE branch_id = ? AND id > ? AND hidden = 0 "
|
|
"AND superseded_by IS NULL ORDER BY id",
|
|
(branch_id, after_id),
|
|
)
|
|
for row in cur:
|
|
yield Event(
|
|
id=row[0], branch_id=row[1], ts=row[2], kind=row[3],
|
|
payload=json.loads(row[4]), superseded_by=row[5], hidden=bool(row[6]),
|
|
)
|