49 lines
1.8 KiB
Python
49 lines
1.8 KiB
Python
|
from datetime import datetime
|
||
|
|
from typing import List, Optional
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends
|
||
|
|
from sqlalchemy import and_, select
|
||
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
|
||
|
|
from ..db import get_session
|
||
|
|
from ..dependencies import AuthUser, get_current_user, require_roles
|
||
|
|
from ..models import AuditAction, AuditLog, AuditResourceType, RoleName
|
||
|
|
from ..schemas import AuditLogOut
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/v1/audit_logs", tags=["audit_logs"])
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("", response_model=List[AuditLogOut])
|
||
|
|
async def list_audit_logs(
|
||
|
|
action: Optional[AuditAction] = None,
|
||
|
|
user_id: Optional[int] = None,
|
||
|
|
customer_id: Optional[int] = None,
|
||
|
|
resource_type: Optional[AuditResourceType] = None,
|
||
|
|
start: Optional[datetime] = None,
|
||
|
|
end: Optional[datetime] = None,
|
||
|
|
limit: int = 50,
|
||
|
|
offset: int = 0,
|
||
|
|
session: AsyncSession = Depends(get_session),
|
||
|
|
auth_user: AuthUser = Depends(require_roles([RoleName.ADMIN, RoleName.CUSTOMER_ADMIN])),
|
||
|
|
) -> List[AuditLogOut]:
|
||
|
|
query = select(AuditLog).order_by(AuditLog.created_at.desc())
|
||
|
|
conditions = []
|
||
|
|
if auth_user.role_name != RoleName.ADMIN.value:
|
||
|
|
conditions.append(AuditLog.customer_id == auth_user.customer_id)
|
||
|
|
if action:
|
||
|
|
conditions.append(AuditLog.action == action)
|
||
|
|
if user_id:
|
||
|
|
conditions.append(AuditLog.user_id == user_id)
|
||
|
|
if customer_id:
|
||
|
|
conditions.append(AuditLog.customer_id == customer_id)
|
||
|
|
if resource_type:
|
||
|
|
conditions.append(AuditLog.resource_type == resource_type)
|
||
|
|
if start:
|
||
|
|
conditions.append(AuditLog.created_at >= start)
|
||
|
|
if end:
|
||
|
|
conditions.append(AuditLog.created_at <= end)
|
||
|
|
if conditions:
|
||
|
|
query = query.where(and_(*conditions))
|
||
|
|
logs = (await session.scalars(query.offset(offset).limit(limit))).all()
|
||
|
|
return [AuditLogOut.model_validate(log) for log in logs]
|