""" audit.py Contiene la lógica de los endpoints utilizados para crear y consultar datos de auditoria. """ from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from sqlalchemy import desc from database import get_db from typing import List, Optional from datetime import datetime import models import schemas from auth import get_current_user router = APIRouter() @router.get("/audit/", response_model=List[schemas.AuditLogResponse], tags=["Audit"]) def get_audit_logs( entity_type: Optional[str] = None, entity_id: Optional[str] = None, action: Optional[str] = None, user_id: Optional[int] = None, date_from: Optional[datetime] = None, date_to: Optional[datetime] = None, skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user: str = Depends(get_current_user), ): """ Consulta el log de auditoría con filtros opcionales. Resultados ordenados del más reciente al más antiguo. """ query = db.query(models.AuditLog).order_by(desc(models.AuditLog.timestamp)) if entity_type: query = query.filter(models.AuditLog.entity_type == entity_type) if entity_id: query = query.filter(models.AuditLog.entity_id == entity_id) if action: query = query.filter(models.AuditLog.action == action) if user_id: query = query.filter(models.AuditLog.user_id == user_id) if date_from: query = query.filter(models.AuditLog.timestamp >= date_from) if date_to: query = query.filter(models.AuditLog.timestamp <= date_to) return query.offset(skip).limit(limit).all() @router.get("/audit/{log_id}", response_model=schemas.AuditLogResponse, tags=["Audit"]) def get_audit_log( log_id: int, db: Session = Depends(get_db), current_user: str = Depends(get_current_user), ): log = db.query(models.AuditLog).filter(models.AuditLog.id == log_id).first() if not log: raise HTTPException(status_code=404, detail="Audit log not found") return log