53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""A simple governance ledger using SQLite for auditability."""
|
|
import sqlite3
|
|
import time
|
|
from typing import List, Dict, Any
|
|
|
|
|
|
class GovernanceLedger:
|
|
def __init__(self, db_path: str = ":memory:"):
|
|
self.conn = sqlite3.connect(db_path)
|
|
self._init_schema()
|
|
|
|
def _init_schema(self) -> None:
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS governance_logs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
action TEXT NOT NULL,
|
|
actor TEXT NOT NULL,
|
|
timestamp REAL NOT NULL,
|
|
status TEXT NOT NULL,
|
|
details TEXT
|
|
)
|
|
"""
|
|
)
|
|
self.conn.commit()
|
|
|
|
def log(self, action: str, actor: str, status: str, details: str | None = None) -> int:
|
|
ts = time.time()
|
|
cur = self.conn.cursor()
|
|
cur.execute(
|
|
"INSERT INTO governance_logs (action, actor, timestamp, status, details) VALUES (?, ?, ?, ?, ?)",
|
|
(action, actor, ts, status, details),
|
|
)
|
|
self.conn.commit()
|
|
return cur.lastrowid
|
|
|
|
def list_logs(self) -> List[Dict[str, Any]]:
|
|
cur = self.conn.cursor()
|
|
cur.execute("SELECT id, action, actor, timestamp, status, details FROM governance_logs ORDER BY timestamp DESC")
|
|
rows = cur.fetchall()
|
|
return [
|
|
{
|
|
"id": r[0],
|
|
"action": r[1],
|
|
"actor": r[2],
|
|
"timestamp": r[3],
|
|
"status": r[4],
|
|
"details": r[5],
|
|
}
|
|
for r in rows
|
|
]
|