"""Audit log viewer router (admin only).""" # Import datetime from datetime from datetime import datetime # Import Optional from typing from typing import Optional # Import APIRouter, Depends, Query from fastapi from fastapi import APIRouter, Depends, Query # Import Session from sqlalchemy.orm from sqlalchemy.orm import Session # Import get_db from app.database from app.database import get_db # Import require_role from app.dependencies.auth from app.dependencies.auth import require_role # Import User from app.models.user from app.models.user import User # Import AuditLogOut, AuditLogPage from app.schemas.audit from app.schemas.audit import AuditLogOut, AuditLogPage # Import from app.services.audit_query_service from app.services.audit_query_service import ( list_distinct_actions, list_distinct_entity_types, list_logs, ) # Assign router = APIRouter(prefix="/audit-logs", tags=["audit"]) router = APIRouter(prefix="/audit-logs", tags=["audit"]) # Apply the @router.get decorator @router.get("", response_model=AuditLogPage) # Define function list_audit_logs def list_audit_logs( # Entry: user_id user_id: Optional[str] = Query(None, description="Filter by user ID"), # Entry: action action: Optional[str] = Query(None, description="Filter by action type"), # Entry: entity_type entity_type: Optional[str] = Query(None, description="Filter by entity type"), # Entry: start_date start_date: Optional[datetime] = Query(None, description="Filter by start date"), # Entry: end_date end_date: Optional[datetime] = Query(None, description="Filter by end date"), # Entry: offset offset: int = Query(0, ge=0, description="Number of records to skip"), # Entry: limit limit: int = Query(50, ge=1, le=100, description="Max records to return"), # Entry: db db: Session = Depends(get_db), # Entry: current_user current_user: User = Depends(require_role("admin")), ) -> AuditLogPage: """Return paginated audit logs with optional filters. **Requires admin role.** """ # Assign result = list_logs( result = list_logs( db, # Keyword argument: user_id user_id=user_id, # Keyword argument: action action=action, # Keyword argument: entity_type entity_type=entity_type, # Keyword argument: start_date start_date=start_date, # Keyword argument: end_date end_date=end_date, # Keyword argument: offset offset=offset, # Keyword argument: limit limit=limit, ) # Return AuditLogPage( return AuditLogPage( # Keyword argument: items items=[AuditLogOut(**item) for item in result["items"]], # Keyword argument: total total=result["total"], # Keyword argument: offset offset=result["offset"], # Keyword argument: limit limit=result["limit"], ) # Apply the @router.get decorator @router.get("/actions", response_model=list[str]) # Define function list_actions def list_actions( # Entry: db db: Session = Depends(get_db), # Entry: current_user current_user: User = Depends(require_role("admin")), ) -> list[str]: """Return a list of distinct action types in the audit log. **Requires admin role.** """ # Return list_distinct_actions(db) return list_distinct_actions(db) # Apply the @router.get decorator @router.get("/entity-types", response_model=list[str]) # Define function list_entity_types def list_entity_types( # Entry: db db: Session = Depends(get_db), # Entry: current_user current_user: User = Depends(require_role("admin")), ) -> list[str]: """Return a list of distinct entity types in the audit log. **Requires admin role.** """ # Return list_distinct_entity_types(db) return list_distinct_entity_types(db)