238 lines
7.9 KiB
Python
238 lines
7.9 KiB
Python
# services/orders/app/api/audit.py
|
|
"""
|
|
Audit Logs API - Retrieve audit trail for orders service
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Path, status
|
|
from typing import Optional, Dict, Any
|
|
from uuid import UUID
|
|
from datetime import datetime
|
|
import structlog
|
|
from sqlalchemy import select, func, and_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import AuditLog
|
|
from shared.auth.decorators import get_current_user_dep
|
|
from shared.auth.access_control import require_user_role
|
|
from shared.routing import RouteBuilder
|
|
from shared.models.audit_log_schemas import (
|
|
AuditLogResponse,
|
|
AuditLogListResponse,
|
|
AuditLogStatsResponse
|
|
)
|
|
from app.core.database import database_manager
|
|
|
|
route_builder = RouteBuilder('orders')
|
|
router = APIRouter(tags=["audit-logs"])
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
async def get_db():
|
|
"""Database session dependency"""
|
|
async with database_manager.get_session() as session:
|
|
yield session
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("audit-logs"),
|
|
response_model=AuditLogListResponse
|
|
)
|
|
@require_user_role(['admin', 'owner'])
|
|
async def get_audit_logs(
|
|
tenant_id: UUID = Path(..., description="Tenant ID"),
|
|
start_date: Optional[datetime] = Query(None, description="Filter logs from this date"),
|
|
end_date: Optional[datetime] = Query(None, description="Filter logs until this date"),
|
|
user_id: Optional[UUID] = Query(None, description="Filter by user ID"),
|
|
action: Optional[str] = Query(None, description="Filter by action type"),
|
|
resource_type: Optional[str] = Query(None, description="Filter by resource type"),
|
|
severity: Optional[str] = Query(None, description="Filter by severity level"),
|
|
search: Optional[str] = Query(None, description="Search in description field"),
|
|
limit: int = Query(100, ge=1, le=1000, description="Number of records to return"),
|
|
offset: int = Query(0, ge=0, description="Number of records to skip"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""
|
|
Get audit logs for orders service.
|
|
Requires admin or owner role.
|
|
"""
|
|
try:
|
|
logger.info(
|
|
"Retrieving audit logs",
|
|
tenant_id=tenant_id,
|
|
user_id=current_user.get("user_id"),
|
|
filters={
|
|
"start_date": start_date,
|
|
"end_date": end_date,
|
|
"action": action,
|
|
"resource_type": resource_type,
|
|
"severity": severity
|
|
}
|
|
)
|
|
|
|
# Build query filters
|
|
filters = [AuditLog.tenant_id == tenant_id]
|
|
|
|
if start_date:
|
|
filters.append(AuditLog.created_at >= start_date)
|
|
if end_date:
|
|
filters.append(AuditLog.created_at <= end_date)
|
|
if user_id:
|
|
filters.append(AuditLog.user_id == user_id)
|
|
if action:
|
|
filters.append(AuditLog.action == action)
|
|
if resource_type:
|
|
filters.append(AuditLog.resource_type == resource_type)
|
|
if severity:
|
|
filters.append(AuditLog.severity == severity)
|
|
if search:
|
|
filters.append(AuditLog.description.ilike(f"%{search}%"))
|
|
|
|
# Count total matching records
|
|
count_query = select(func.count()).select_from(AuditLog).where(and_(*filters))
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
# Fetch paginated results
|
|
query = (
|
|
select(AuditLog)
|
|
.where(and_(*filters))
|
|
.order_by(AuditLog.created_at.desc())
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
|
|
result = await db.execute(query)
|
|
audit_logs = result.scalars().all()
|
|
|
|
# Convert to response models
|
|
items = [AuditLogResponse.from_orm(log) for log in audit_logs]
|
|
|
|
logger.info(
|
|
"Successfully retrieved audit logs",
|
|
tenant_id=tenant_id,
|
|
total=total,
|
|
returned=len(items)
|
|
)
|
|
|
|
return AuditLogListResponse(
|
|
items=items,
|
|
total=total,
|
|
limit=limit,
|
|
offset=offset,
|
|
has_more=(offset + len(items)) < total
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
"Failed to retrieve audit logs",
|
|
error=str(e),
|
|
tenant_id=tenant_id
|
|
)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to retrieve audit logs: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("audit-logs/stats"),
|
|
response_model=AuditLogStatsResponse
|
|
)
|
|
@require_user_role(['admin', 'owner'])
|
|
async def get_audit_log_stats(
|
|
tenant_id: UUID = Path(..., description="Tenant ID"),
|
|
start_date: Optional[datetime] = Query(None, description="Filter logs from this date"),
|
|
end_date: Optional[datetime] = Query(None, description="Filter logs until this date"),
|
|
current_user: Dict[str, Any] = Depends(get_current_user_dep),
|
|
db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""
|
|
Get audit log statistics for orders service.
|
|
Requires admin or owner role.
|
|
"""
|
|
try:
|
|
logger.info(
|
|
"Retrieving audit log statistics",
|
|
tenant_id=tenant_id,
|
|
user_id=current_user.get("user_id")
|
|
)
|
|
|
|
# Build base filters
|
|
filters = [AuditLog.tenant_id == tenant_id]
|
|
if start_date:
|
|
filters.append(AuditLog.created_at >= start_date)
|
|
if end_date:
|
|
filters.append(AuditLog.created_at <= end_date)
|
|
|
|
# Total events
|
|
count_query = select(func.count()).select_from(AuditLog).where(and_(*filters))
|
|
total_result = await db.execute(count_query)
|
|
total_events = total_result.scalar() or 0
|
|
|
|
# Events by action
|
|
action_query = (
|
|
select(AuditLog.action, func.count().label('count'))
|
|
.where(and_(*filters))
|
|
.group_by(AuditLog.action)
|
|
)
|
|
action_result = await db.execute(action_query)
|
|
events_by_action = {row.action: row.count for row in action_result}
|
|
|
|
# Events by severity
|
|
severity_query = (
|
|
select(AuditLog.severity, func.count().label('count'))
|
|
.where(and_(*filters))
|
|
.group_by(AuditLog.severity)
|
|
)
|
|
severity_result = await db.execute(severity_query)
|
|
events_by_severity = {row.severity: row.count for row in severity_result}
|
|
|
|
# Events by resource type
|
|
resource_query = (
|
|
select(AuditLog.resource_type, func.count().label('count'))
|
|
.where(and_(*filters))
|
|
.group_by(AuditLog.resource_type)
|
|
)
|
|
resource_result = await db.execute(resource_query)
|
|
events_by_resource_type = {row.resource_type: row.count for row in resource_result}
|
|
|
|
# Date range
|
|
date_range_query = (
|
|
select(
|
|
func.min(AuditLog.created_at).label('min_date'),
|
|
func.max(AuditLog.created_at).label('max_date')
|
|
)
|
|
.where(and_(*filters))
|
|
)
|
|
date_result = await db.execute(date_range_query)
|
|
date_row = date_result.one()
|
|
|
|
logger.info(
|
|
"Successfully retrieved audit log statistics",
|
|
tenant_id=tenant_id,
|
|
total_events=total_events
|
|
)
|
|
|
|
return AuditLogStatsResponse(
|
|
total_events=total_events,
|
|
events_by_action=events_by_action,
|
|
events_by_severity=events_by_severity,
|
|
events_by_resource_type=events_by_resource_type,
|
|
date_range={
|
|
"min": date_row.min_date,
|
|
"max": date_row.max_date
|
|
}
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
"Failed to retrieve audit log statistics",
|
|
error=str(e),
|
|
tenant_id=tenant_id
|
|
)
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to retrieve audit log statistics: {str(e)}"
|
|
)
|