diff --git a/app/routes/audit.py b/app/routes/audit.py new file mode 100644 index 0000000..60e722e --- /dev/null +++ b/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, + }