NuklearRabbit
7e34f55005
audit_events already stored before_json/after_json, but the API and UI only ever surfaced metadata -- the audit trail could say something happened but never show what changed. Add before/after to AuditEventOut, resolve a safe entity_ref/entity_link for vehicle/booking/data-quality-issue entities (customer stays label-only; no customer detail route exists in this PoC), and render a human-readable change summary in the UI with the raw before/after/metadata JSON kept behind a <details> disclosure rather than shown by default.
98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from collections.abc import Sequence
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_db, require_operations_manager
|
|
from app.models.audit import AuditEvent
|
|
from app.models.booking import Booking
|
|
from app.models.customer import Customer
|
|
from app.models.data_quality import DataQualityIssue
|
|
from app.models.vehicle import Vehicle
|
|
from app.schemas import AuditEventOut, CurrentUser
|
|
|
|
router = APIRouter(prefix="/api/v1/audit", tags=["audit"])
|
|
|
|
# Only entity types with a stable public reference and (optionally) a real frontend route
|
|
# are resolved here. Types like "system", "knowledge" or "mcp_tool" carry no linkable
|
|
# entity_id and are left as plain labels.
|
|
_ENTITY_MODELS: dict[str, Any] = {
|
|
"vehicle": Vehicle,
|
|
"booking": Booking,
|
|
"customer": Customer,
|
|
"data_quality_issue": DataQualityIssue,
|
|
}
|
|
_ROUTE_TEMPLATES: dict[str, str] = {
|
|
"vehicle": "/vehicles/{ref}",
|
|
"booking": "/bookings/{ref}",
|
|
"data_quality_issue": "/data-quality/{ref}",
|
|
# No customer detail route exists in this proof of concept; still resolve the
|
|
# reference for display, just without a link.
|
|
}
|
|
|
|
|
|
def _resolve_entity_refs(db: Session, events: Sequence[AuditEvent]) -> dict[uuid.UUID, str]:
|
|
ids_by_type: dict[str, set[uuid.UUID]] = {}
|
|
for event in events:
|
|
if event.entity_id is not None and event.entity_type in _ENTITY_MODELS:
|
|
ids_by_type.setdefault(event.entity_type, set()).add(event.entity_id)
|
|
|
|
refs: dict[uuid.UUID, str] = {}
|
|
for entity_type, ids in ids_by_type.items():
|
|
model = _ENTITY_MODELS[entity_type]
|
|
rows: Sequence[Any] = db.scalars(select(model).where(model.id.in_(ids))).all()
|
|
for row in rows:
|
|
refs[row.id] = row.public_ref
|
|
return refs
|
|
|
|
|
|
@router.get("", response_model=list[AuditEventOut])
|
|
def list_audit_events(
|
|
actor_label: str | None = Query(default=None),
|
|
action: str | None = Query(default=None),
|
|
entity_type: str | None = Query(default=None),
|
|
correlation_id: str | None = Query(default=None),
|
|
limit: int = Query(default=100, le=500),
|
|
db: Session = Depends(get_db),
|
|
_user: CurrentUser = Depends(require_operations_manager),
|
|
) -> list[AuditEventOut]:
|
|
stmt = select(AuditEvent).order_by(AuditEvent.occurred_at.desc()).limit(limit)
|
|
if actor_label:
|
|
stmt = stmt.where(AuditEvent.actor_label == actor_label)
|
|
if action:
|
|
stmt = stmt.where(AuditEvent.action == action)
|
|
if entity_type:
|
|
stmt = stmt.where(AuditEvent.entity_type == entity_type)
|
|
if correlation_id:
|
|
stmt = stmt.where(AuditEvent.correlation_id == correlation_id)
|
|
events = db.scalars(stmt).all()
|
|
entity_refs = _resolve_entity_refs(db, events)
|
|
|
|
out = []
|
|
for e in events:
|
|
ref = entity_refs.get(e.entity_id) if e.entity_id else None
|
|
route = _ROUTE_TEMPLATES.get(e.entity_type)
|
|
out.append(
|
|
AuditEventOut(
|
|
id=str(e.id),
|
|
actor_type=e.actor_type,
|
|
actor_label=e.actor_label,
|
|
action=e.action,
|
|
entity_type=e.entity_type,
|
|
entity_id=str(e.entity_id) if e.entity_id else None,
|
|
entity_ref=ref,
|
|
entity_link=route.format(ref=ref) if route and ref else None,
|
|
correlation_id=str(e.correlation_id),
|
|
occurred_at=e.occurred_at,
|
|
before=e.before_json,
|
|
after=e.after_json,
|
|
metadata=e.metadata_json,
|
|
)
|
|
)
|
|
return out
|