NuklearRabbit
760f3b6ee2
The data-quality workbench (list, detail, defer, reject) and the audit trail had no role gate at all beyond authentication -- confirmed live, a Rental Employee session could list and resolve data-quality issues and read the full audit trail through both the API and the UI, with only merge-customers and scan already restricted. Per the role matrix, both areas are Operations-Manager-only. Gate the remaining data-quality and audit endpoints with require_operations_manager, hide their nav items for Rental Employee, show the same restricted-message pattern Automation.tsx already used for direct URL access, and stop the dashboard from linking into now-restricted areas for that role.
48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
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.schemas import AuditEventOut, CurrentUser
|
|
|
|
router = APIRouter(prefix="/api/v1/audit", tags=["audit"])
|
|
|
|
|
|
@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()
|
|
return [
|
|
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,
|
|
correlation_id=str(e.correlation_id),
|
|
occurred_at=e.occurred_at,
|
|
metadata=e.metadata_json,
|
|
)
|
|
for e in events
|
|
]
|