C6: Add audit log route — admin-only, paginated, filterable

app/routes/audit.py

@@ -0,0 +1,99 @@
+"""Audit log routes — admin-only, paginated, filterable."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.deps import get_current_user
+from app.models.audit import AuditLog
+
+router = APIRouter(prefix="/api/v1/audit-log", tags=["audit"])
+
+
+def _audit_to_dict(a: AuditLog) -> dict:
+    """Serialize an AuditLog ORM object to dict."""
+    return {
+        "id": str(a.id),
+        "user_id": str(a.user_id) if a.user_id else None,
+        "action": a.action,
+        "entity_type": a.entity_type,
+        "entity_id": str(a.entity_id) if a.entity_id else None,
+        "changes": a.changes,
+        "timestamp": a.timestamp.isoformat() if a.timestamp else None,
+        "tenant_id": str(a.tenant_id),
+    }
+
+
+@router.get("")
+async def list_audit_logs(
+    entity_type: str | None = Query(None, description="Filter by entity type"),
+    user_id: str | None = Query(None, description="Filter by user ID"),
+    action: str | None = Query(None, description="Filter by action (create/update/delete/login)"),
+    date_from: str | None = Query(None, description="ISO date range start (inclusive)"),
+    date_to: str | None = Query(None, description="ISO date range end (inclusive)"),
+    page: int = Query(1, ge=1, description="Page number"),
+    page_size: int = Query(50, ge=1, le=200, description="Items per page"),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(get_current_user),
+):
+    """List audit log entries with filtering and pagination. Admin only."""
+    if current_user.get("role") != "admin":
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={"detail": "Admin access required", "code": "forbidden"},
+        )
+
+    tenant_id = uuid.UUID(current_user["tenant_id"])
+
+    q = select(AuditLog).where(
+        AuditLog.tenant_id == tenant_id,
+        AuditLog.deleted_at.is_(None),
+    )
+
+    if entity_type:
+        q = q.where(AuditLog.entity_type == entity_type)
+    if user_id:
+        try:
+            uid = uuid.UUID(user_id)
+        except ValueError:
+            raise HTTPException(400, detail={"detail": "Invalid user_id", "code": "invalid_id"}) from None
+        q = q.where(AuditLog.user_id == uid)
+    if action:
+        q = q.where(AuditLog.action == action)
+    if date_from:
+        try:
+            dt_from = datetime.fromisoformat(date_from)
+        except ValueError:
+            raise HTTPException(400, detail={"detail": "Invalid date_from format", "code": "invalid_date"}) from None
+        q = q.where(AuditLog.timestamp >= dt_from)
+    if date_to:
+        try:
+            dt_to = datetime.fromisoformat(date_to)
+        except ValueError:
+            raise HTTPException(400, detail={"detail": "Invalid date_to format", "code": "invalid_date"}) from None
+        q = q.where(AuditLog.timestamp <= dt_to)
+
+    # Count total
+    count_q = select(func.count()).select_from(q.subquery())
+    count_result = await db.execute(count_q)
+    total = count_result.scalar() or 0
+
+    # Apply pagination
+    offset = (page - 1) * page_size
+    q = q.order_by(AuditLog.timestamp.desc()).offset(offset).limit(page_size)
+    result = await db.execute(q)
+    entries = result.scalars().all()
+
+    return {
+        "items": [_audit_to_dict(a) for a in entries],
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+        "total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0,
+    }