42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
from datetime import datetime
|
|
from typing import List, Optional
|
|
|
|
from sqlalchemy import and_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from backend.modules.audit.models import AuditAction, AuditLog, AuditResourceType
|
|
|
|
|
|
async def list_audit_logs(
|
|
session: AsyncSession,
|
|
customer_id: Optional[int],
|
|
user_id: Optional[int],
|
|
action: Optional[AuditAction],
|
|
start: Optional[datetime],
|
|
end: Optional[datetime],
|
|
) -> List[AuditLog]:
|
|
query = (
|
|
select(AuditLog)
|
|
.options(selectinload(AuditLog.user), selectinload(AuditLog.customer))
|
|
.order_by(AuditLog.created_at.desc())
|
|
)
|
|
conditions = []
|
|
if customer_id:
|
|
conditions.append(AuditLog.customer_id == customer_id)
|
|
if user_id:
|
|
conditions.append(AuditLog.user_id == user_id)
|
|
if action:
|
|
conditions.append(AuditLog.action == action)
|
|
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)).all()
|
|
for log in logs:
|
|
setattr(log, "user_name", log.user.username if log.user else None)
|
|
setattr(log, "customer_name", log.customer.name if log.customer else None)
|
|
return logs
|