49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
import json
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.auth import get_current_admin
|
|
from app.database import get_db
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.user import User
|
|
from app.schemas import AuditLogResponse
|
|
|
|
router = APIRouter(prefix="/api/logs", tags=["logs"])
|
|
|
|
|
|
def _log_to_response(log: AuditLog) -> AuditLogResponse:
|
|
details = None
|
|
if log.details:
|
|
try:
|
|
details = json.loads(log.details)
|
|
except json.JSONDecodeError:
|
|
details = {"raw": log.details}
|
|
return AuditLogResponse(
|
|
id=log.id,
|
|
user_id=log.user_id,
|
|
username=log.username,
|
|
action=log.action,
|
|
resource_type=log.resource_type,
|
|
resource_id=log.resource_id,
|
|
details=details,
|
|
ip_address=log.ip_address,
|
|
created_at=log.created_at,
|
|
)
|
|
|
|
|
|
@router.get("", response_model=list[AuditLogResponse])
|
|
def list_logs(
|
|
db: Annotated[Session, Depends(get_db)],
|
|
_: Annotated[User, Depends(get_current_admin)],
|
|
limit: int = Query(default=200, ge=1, le=1000),
|
|
offset: int = Query(default=0, ge=0),
|
|
action: str | None = None,
|
|
):
|
|
query = db.query(AuditLog).order_by(AuditLog.created_at.desc())
|
|
if action:
|
|
query = query.filter(AuditLog.action.contains(action))
|
|
logs = query.offset(offset).limit(limit).all()
|
|
return [_log_to_response(log) for log in logs]
|