JiriUhlir
78 lines
1.5 KiB
Python
78 lines
1.5 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
from app.db.database import get_connection
|
|
|
|
|
|
def log_audit_event(
|
|
user,
|
|
action: str,
|
|
target_type: str,
|
|
target_id=None,
|
|
source: str = "portal",
|
|
metadata: dict[str, Any] | None = None,
|
|
):
|
|
if user:
|
|
user_id = user.get("id")
|
|
username = user.get("username") or "system"
|
|
else:
|
|
user_id = None
|
|
username = "system"
|
|
source = "system"
|
|
|
|
metadata_json = json.dumps(metadata or {}, ensure_ascii=False, sort_keys=True)
|
|
|
|
con = get_connection()
|
|
con.execute(
|
|
"""
|
|
INSERT INTO audit_events (
|
|
user_id,
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
|
""",
|
|
(
|
|
user_id,
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata_json,
|
|
),
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def get_audit_events(limit: int = 200):
|
|
con = get_connection()
|
|
|
|
rows = con.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
user_id,
|
|
username,
|
|
action,
|
|
target_type,
|
|
target_id,
|
|
source,
|
|
metadata,
|
|
created_at
|
|
FROM audit_events
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
|
|
con.close()
|
|
return [dict(row) for row in rows]
|