68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from app.core.contracts import PermissionDecision
|
|
|
|
|
|
class SQLiteApprovalStore:
|
|
"""Stores persistent user approval decisions."""
|
|
|
|
def __init__(self, db_path: str | Path) -> None:
|
|
self._db_path = Path(db_path)
|
|
self._db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
|
|
def save(self, decision: PermissionDecision) -> PermissionDecision:
|
|
with sqlite3.connect(self._db_path) as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT OR REPLACE INTO approvals (action_type, pattern, decision, created_at)
|
|
VALUES (?, ?, ?, ?)
|
|
""",
|
|
(
|
|
decision.action_type,
|
|
decision.pattern,
|
|
decision.decision,
|
|
decision.created_at.isoformat(),
|
|
),
|
|
)
|
|
conn.commit()
|
|
return decision
|
|
|
|
def load(self, action_type: str, pattern: str) -> PermissionDecision | None:
|
|
with sqlite3.connect(self._db_path) as conn:
|
|
row = conn.execute(
|
|
"""
|
|
SELECT action_type, pattern, decision, created_at
|
|
FROM approvals
|
|
WHERE action_type = ? AND pattern = ?
|
|
""",
|
|
(action_type, pattern),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
return PermissionDecision(
|
|
action_type=row[0],
|
|
pattern=row[1],
|
|
decision=row[2],
|
|
created_at=row[3],
|
|
)
|
|
|
|
def _initialize(self) -> None:
|
|
with sqlite3.connect(self._db_path) as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS approvals (
|
|
action_type TEXT NOT NULL,
|
|
pattern TEXT NOT NULL,
|
|
decision TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
PRIMARY KEY (action_type, pattern)
|
|
)
|
|
"""
|
|
)
|
|
conn.commit()
|
|
|