src/db/audit.py

import sqlite3
from typing import Optional, List, Dict
from src.utils.time_utils import utc_now_iso
from src.utils.sqlite_retry import run_tx

def log(conn: sqlite3.Connection, entity: str, entity_id: Optional[str], action: str, note: str, changed_by: str) -> None:
    now = utc_now_iso()
    def _tx(c):
        c.execute(
            "INSERT INTO audit_log(entity, entity_id, action, note, changed_by, changed_at_utc) VALUES(?,?,?,?,?,?)",
            (entity, entity_id, action, note, changed_by, now),
        )
    run_tx(conn, _tx)

def recent(conn: sqlite3.Connection, limit: int = 20) -> List[Dict]:
    rows = conn.execute(
        "SELECT entity, entity_id, action, note, changed_by, changed_at_utc FROM audit_log ORDER BY id DESC LIMIT ?",
        (limit,),
    ).fetchall()
    return [dict(r) for r in rows]