126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
"""Журнал аудита «кто/когда/что/old->new» (issue #20).
|
|
|
|
Запись всегда выполняется на переданном соединении, то есть внутри уже
|
|
открытой транзакции вызывающего кода: мутация и её аудит коммитятся вместе и
|
|
не могут разойтись.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
from .clock import utcnow
|
|
from .models import AUDIT_STATUS_OK, AuditEntry
|
|
|
|
__all__ = ["list_entries", "record"]
|
|
|
|
|
|
def _dumps(value: Any) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, Mapping):
|
|
return json.dumps(dict(value), sort_keys=True, ensure_ascii=False, default=str)
|
|
return json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)
|
|
|
|
|
|
def _loads(value: Any) -> Any:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (bytes, bytearray)):
|
|
value = value.decode("utf-8")
|
|
return json.loads(str(value))
|
|
|
|
|
|
def record(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
actor: str,
|
|
action: str,
|
|
entity: str,
|
|
entity_id: str | int | None = None,
|
|
old_value: Any = None,
|
|
new_value: Any = None,
|
|
status: str = AUDIT_STATUS_OK,
|
|
detail: str | None = None,
|
|
ts: str | None = None,
|
|
) -> int:
|
|
"""Записать событие в ``audit_log`` и вернуть id записи."""
|
|
if not actor:
|
|
raise ValueError("actor обязателен: аудит должен знать, кто изменил состояние")
|
|
cursor = conn.execute(
|
|
"""
|
|
INSERT INTO audit_log (ts, actor, action, entity, entity_id,
|
|
old_value, new_value, status, detail)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
ts or utcnow(),
|
|
actor,
|
|
action,
|
|
entity,
|
|
None if entity_id is None else str(entity_id),
|
|
_dumps(old_value),
|
|
_dumps(new_value),
|
|
status,
|
|
detail,
|
|
),
|
|
)
|
|
return int(cursor.lastrowid or 0)
|
|
|
|
|
|
def list_entries(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
entity: str | None = None,
|
|
entity_id: str | int | None = None,
|
|
limit: int = 100,
|
|
) -> list[AuditEntry]:
|
|
"""Прочитать журнал аудита (свежие первыми)."""
|
|
sql = "SELECT * FROM audit_log"
|
|
clauses: list[str] = []
|
|
params: list[Any] = []
|
|
if entity is not None:
|
|
clauses.append("entity = ?")
|
|
params.append(entity)
|
|
if entity_id is not None:
|
|
clauses.append("entity_id = ?")
|
|
params.append(str(entity_id))
|
|
if clauses:
|
|
sql += " WHERE " + " AND ".join(clauses)
|
|
sql += " ORDER BY id DESC LIMIT ?"
|
|
params.append(int(limit))
|
|
rows = conn.execute(sql, params).fetchall()
|
|
return [
|
|
AuditEntry(
|
|
id=int(row["id"]),
|
|
ts=str(row["ts"]),
|
|
actor=str(row["actor"]),
|
|
action=str(row["action"]),
|
|
entity=str(row["entity"]),
|
|
entity_id=None if row["entity_id"] is None else str(row["entity_id"]),
|
|
old_value=_loads(row["old_value"]),
|
|
new_value=_loads(row["new_value"]),
|
|
status=str(row["status"]),
|
|
detail=None if row["detail"] is None else str(row["detail"]),
|
|
)
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def count(conn: sqlite3.Connection, *, entity: str | None = None) -> int:
|
|
"""Сколько записей в журнале (опционально — по сущности)."""
|
|
if entity is None:
|
|
row = conn.execute("SELECT COUNT(*) FROM audit_log").fetchone()
|
|
else:
|
|
row = conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity = ?", (entity,)).fetchone()
|
|
return int(row[0]) if row is not None else 0
|
|
|
|
|
|
def last_entry(conn: sqlite3.Connection, *, entity: str | None = None) -> AuditEntry | None:
|
|
"""Последняя запись журнала (опционально — по сущности)."""
|
|
entries = list_entries(conn, entity=entity, limit=1)
|
|
return entries[0] if entries else None
|