File size: 2,057 Bytes
c9dcc3b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | from __future__ import annotations
import json
import sqlite3
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from triage_llm.utils import ensure_dir
class AuditStore:
def __init__(self, db_path: str = "runs/audit/audit.db") -> None:
self.db_path = Path(db_path)
ensure_dir(self.db_path.parent)
self._init_db()
def _init_db(self) -> None:
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS interactions (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
request_json TEXT NOT NULL,
response_json TEXT NOT NULL
)
"""
)
def save(self, request: dict[str, Any], response: dict[str, Any]) -> str:
interaction_id = str(uuid.uuid4())
created_at = datetime.now(timezone.utc).isoformat()
with sqlite3.connect(self.db_path) as conn:
conn.execute(
(
"INSERT INTO interactions (id, created_at, request_json, response_json) "
"VALUES (?, ?, ?, ?)"
),
(
interaction_id,
created_at,
json.dumps(request, ensure_ascii=False),
json.dumps(response, ensure_ascii=False),
),
)
return interaction_id
def get(self, interaction_id: str) -> dict[str, Any] | None:
with sqlite3.connect(self.db_path) as conn:
row = conn.execute(
"SELECT id, created_at, request_json, response_json FROM interactions WHERE id = ?",
(interaction_id,),
).fetchone()
if row is None:
return None
return {
"id": row[0],
"created_at": row[1],
"request": json.loads(row[2]),
"response": json.loads(row[3]),
}
|