133 lines
5.1 KiB
Python
133 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import uuid
|
|
from collections.abc import Sequence
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, 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, AuditEventPageOut, 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] | AuditEventPageOut)
|
|
def list_audit_events(
|
|
actor_label: str | None = Query(default=None),
|
|
action: str | None = Query(default=None),
|
|
entity_type: str | None = Query(default=None),
|
|
entity_ref: str | None = Query(default=None, min_length=1, max_length=100),
|
|
correlation_id: str | None = Query(default=None),
|
|
occurred_from: datetime | None = Query(default=None),
|
|
occurred_to: datetime | None = Query(default=None),
|
|
page: int | None = Query(default=None, ge=1),
|
|
page_size: int = Query(default=25, ge=1, le=25),
|
|
db: Session = Depends(get_db),
|
|
_user: CurrentUser = Depends(require_operations_manager),
|
|
) -> list[AuditEventOut] | AuditEventPageOut:
|
|
stmt = select(AuditEvent).order_by(AuditEvent.occurred_at.desc())
|
|
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 entity_ref:
|
|
matched_ids: set[uuid.UUID] = set()
|
|
for model in _ENTITY_MODELS.values():
|
|
matched_ids.update(
|
|
db.scalars(select(model.id).where(model.public_ref.ilike(f"%{entity_ref.strip()}%"))).all()
|
|
)
|
|
if not matched_ids:
|
|
if page is None:
|
|
return []
|
|
return AuditEventPageOut(
|
|
items=[], page=1, page_size=page_size, total=0, total_pages=1
|
|
)
|
|
stmt = stmt.where(AuditEvent.entity_id.in_(matched_ids))
|
|
if correlation_id:
|
|
stmt = stmt.where(AuditEvent.correlation_id == correlation_id)
|
|
if occurred_from:
|
|
stmt = stmt.where(AuditEvent.occurred_at >= occurred_from)
|
|
if occurred_to:
|
|
stmt = stmt.where(AuditEvent.occurred_at <= occurred_to)
|
|
total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0
|
|
page_number = page or 1
|
|
events = db.scalars(
|
|
stmt if page is None else stmt.offset((page_number - 1) * page_size).limit(page_size)
|
|
).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,
|
|
)
|
|
)
|
|
if page is None:
|
|
return out
|
|
total_pages = max(1, (total + page_size - 1) // page_size)
|
|
return AuditEventPageOut(
|
|
items=out,
|
|
page=min(page_number, total_pages),
|
|
page_size=page_size,
|
|
total=total,
|
|
total_pages=total_pages,
|
|
)
|