50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
from fastapi import APIRouter, Depends, Query, status
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import require_roles
|
|
from app.core.errors import raise_api_error
|
|
from app.db.session import get_db
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.user import User
|
|
from app.schemas.audit import AuditLogRead
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=list[AuditLogRead])
|
|
def list_audit_logs(
|
|
_: User = Depends(require_roles("admin")),
|
|
db: Session = Depends(get_db),
|
|
actor_id: int | None = Query(default=None),
|
|
target_type: str | None = Query(default=None),
|
|
action_name: str | None = Query(default=None, alias="action"),
|
|
limit: int = Query(default=100, ge=1, le=300),
|
|
) -> list[AuditLog]:
|
|
query = select(AuditLog)
|
|
if actor_id is not None:
|
|
query = query.where(AuditLog.actor_id == actor_id)
|
|
if target_type is not None:
|
|
query = query.where(AuditLog.target_type == target_type)
|
|
if action_name is not None:
|
|
query = query.where(AuditLog.action == action_name)
|
|
query = query.order_by(AuditLog.id.desc()).limit(limit)
|
|
return list(db.scalars(query))
|
|
|
|
|
|
@router.get("/{audit_id}", response_model=AuditLogRead)
|
|
def get_audit_log(
|
|
audit_id: int,
|
|
_: User = Depends(require_roles("admin")),
|
|
db: Session = Depends(get_db),
|
|
) -> AuditLog:
|
|
audit_log = db.get(AuditLog, audit_id)
|
|
if audit_log is None:
|
|
raise_api_error(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
code="audit_not_found",
|
|
message="Requested audit record was not found.",
|
|
)
|
|
return audit_log
|