85 lines
2.4 KiB
Python
85 lines
2.4 KiB
Python
"""
|
|
Audit log utility — all services call log_action() to record staff events.
|
|
|
|
Usage:
|
|
from shared.audit import log_action
|
|
|
|
await log_action(
|
|
db, actor_id, actor_name,
|
|
action="CREATE", entity_type="customer", entity_id=cust_id,
|
|
entity_label="Church of St. George",
|
|
)
|
|
|
|
await log_action(
|
|
db, actor_id, actor_name,
|
|
action="UPDATE", entity_type="order", entity_id=order_id,
|
|
entity_label="ORD-0042",
|
|
changes={"status": {"old": "negotiating", "new": "confirmed"}},
|
|
)
|
|
|
|
Never raises — a logging failure must never break the primary operation.
|
|
The call is fire-and-forget safe: wrap in try/except internally.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from shared.orm import AuditLog
|
|
|
|
|
|
async def log_action(
|
|
db: AsyncSession,
|
|
actor_id: str,
|
|
actor_name: str,
|
|
action: str,
|
|
entity_type: str,
|
|
entity_id: str,
|
|
entity_label: str | None = None,
|
|
changes: dict[str, Any] | None = None,
|
|
meta: dict[str, Any] | None = None,
|
|
) -> None:
|
|
"""
|
|
Insert one row into audit_log. Never raises — failures are silently swallowed
|
|
so a logging error never disrupts the primary request.
|
|
|
|
Always commits its own mini-transaction. Callers that run inside a larger
|
|
transaction (e.g. staff service) should commit themselves after calling this;
|
|
the extra commit here is a no-op if the session is already clean.
|
|
"""
|
|
try:
|
|
entry = AuditLog(
|
|
occurred_at=datetime.now(timezone.utc),
|
|
actor_id=actor_id,
|
|
actor_name=actor_name,
|
|
action=action,
|
|
entity_type=entity_type,
|
|
entity_id=entity_id,
|
|
entity_label=entity_label,
|
|
changes=changes,
|
|
meta=meta,
|
|
)
|
|
db.add(entry)
|
|
await db.commit()
|
|
except Exception:
|
|
await db.rollback()
|
|
|
|
|
|
def diff(old: dict, new: dict) -> dict[str, dict]:
|
|
"""
|
|
Build a changes dict from two flat dicts.
|
|
Only includes keys whose values actually changed.
|
|
Skip internal/unloggable keys (hashed_password, updated_at).
|
|
|
|
Usage:
|
|
changes = diff(old_record, new_record)
|
|
await log_action(..., changes=changes or None)
|
|
"""
|
|
_SKIP = {"hashed_password", "updated_at", "firestore_id"}
|
|
return {
|
|
k: {"old": old.get(k), "new": new.get(k)}
|
|
for k in new
|
|
if k not in _SKIP and old.get(k) != new.get(k)
|
|
}
|