feat: punkt 7 (audit log) — export route (CSV/JSON), retention cleanup ARQ cron job (daily 03:00, 365 days default)
This commit is contained in:
@@ -342,6 +342,49 @@ async def cleanup_outbox_job(ctx: dict[str, Any]) -> None:
|
|||||||
register_job("cleanup_outbox", cleanup_outbox_job)
|
register_job("cleanup_outbox", cleanup_outbox_job)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Audit log retention cleanup job ─────────────────────────────────────────
|
||||||
|
|
||||||
|
async def cleanup_audit_log_job(ctx: dict[str, Any]) -> None:
|
||||||
|
"""Delete audit log entries older than 365 days.
|
||||||
|
|
||||||
|
Runs daily to prevent the audit_log table from growing indefinitely.
|
||||||
|
Iterates per-tenant for RLS compliance.
|
||||||
|
"""
|
||||||
|
from sqlalchemy import text as sa_text, delete as sa_delete
|
||||||
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
|
from app.core.db import get_worker_session_factory
|
||||||
|
from app.models.audit import AuditLog
|
||||||
|
|
||||||
|
factory = get_worker_session_factory()
|
||||||
|
async with factory() as db:
|
||||||
|
try:
|
||||||
|
tenant_result = await db.execute(sa_text("SELECT id FROM tenants"))
|
||||||
|
tenant_ids = [row[0] for row in tenant_result]
|
||||||
|
|
||||||
|
cutoff = datetime.utcnow() - timedelta(days=365)
|
||||||
|
total_deleted = 0
|
||||||
|
for tenant_id in tenant_ids:
|
||||||
|
await db.execute(
|
||||||
|
sa_text("SELECT set_config('app.current_tenant_id', :tid, true)"),
|
||||||
|
{"tid": str(tenant_id)},
|
||||||
|
)
|
||||||
|
result = await db.execute(
|
||||||
|
sa_delete(AuditLog).where(AuditLog.timestamp < cutoff)
|
||||||
|
)
|
||||||
|
total_deleted += result.rowcount
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
if total_deleted:
|
||||||
|
logger.info("Audit retention: cleaned up %d old entries", total_deleted)
|
||||||
|
except Exception:
|
||||||
|
logger.error("Audit retention cleanup failed", exc_info=True)
|
||||||
|
await db.rollback()
|
||||||
|
|
||||||
|
|
||||||
|
register_job("cleanup_audit_log", cleanup_audit_log_job)
|
||||||
|
|
||||||
|
|
||||||
class WorkerSettings:
|
class WorkerSettings:
|
||||||
"""ARQ worker settings."""
|
"""ARQ worker settings."""
|
||||||
functions = get_all_jobs()
|
functions = get_all_jobs()
|
||||||
@@ -371,6 +414,11 @@ class WorkerSettings:
|
|||||||
_wrap_cron_with_lock("cleanup_outbox", cleanup_outbox_job, ttl_seconds=300),
|
_wrap_cron_with_lock("cleanup_outbox", cleanup_outbox_job, ttl_seconds=300),
|
||||||
minute=0,
|
minute=0,
|
||||||
),
|
),
|
||||||
|
# Audit log retention cleanup — daily at 03:00
|
||||||
|
cron(
|
||||||
|
_wrap_cron_with_lock("cleanup_audit_log", cleanup_audit_log_job, ttl_seconds=300),
|
||||||
|
hour=3, minute=0,
|
||||||
|
),
|
||||||
# Scheduled backup — daily at 02:00 (guarded by distributed lock)
|
# Scheduled backup — daily at 02:00 (guarded by distributed lock)
|
||||||
cron(
|
cron(
|
||||||
_wrap_cron_with_lock("run_backup", get_job("run_backup"), ttl_seconds=600),
|
_wrap_cron_with_lock("run_backup", get_job("run_backup"), ttl_seconds=600),
|
||||||
|
|||||||
+88
-3
@@ -1,12 +1,16 @@
|
|||||||
"""Audit log routes — admin-only, paginated, filterable."""
|
"""Audit log routes — admin-only, paginated, filterable, export, retention."""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import csv
|
||||||
|
import io
|
||||||
|
import json
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import datetime
|
from datetime import datetime, timedelta
|
||||||
|
|
||||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||||
from sqlalchemy import func, select
|
from fastapi.responses import StreamingResponse
|
||||||
|
from sqlalchemy import func, select, delete
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from app.core.db import get_db
|
from app.core.db import get_db
|
||||||
@@ -92,3 +96,84 @@ async def list_audit_logs(
|
|||||||
"page_size": page_size,
|
"page_size": page_size,
|
||||||
"total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0,
|
"total_pages": (total + page_size - 1) // page_size if page_size > 0 else 0,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/export")
|
||||||
|
async def export_audit_logs(
|
||||||
|
format: str = Query("csv", description="Export format: csv or json"),
|
||||||
|
entity_type: str | None = Query(None),
|
||||||
|
action: str | None = Query(None),
|
||||||
|
date_from: str | None = Query(None),
|
||||||
|
date_to: str | None = Query(None),
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
|
current_user: dict = Depends(require_permission("audit:read")),
|
||||||
|
):
|
||||||
|
"""Export audit log entries as CSV or JSON. Admin only."""
|
||||||
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||||
|
|
||||||
|
q = select(AuditLog).where(
|
||||||
|
AuditLog.tenant_id == tenant_id,
|
||||||
|
AuditLog.deleted_at.is_(None),
|
||||||
|
)
|
||||||
|
if entity_type:
|
||||||
|
q = q.where(AuditLog.entity_type == entity_type)
|
||||||
|
if action:
|
||||||
|
q = q.where(AuditLog.action == action)
|
||||||
|
if date_from:
|
||||||
|
q = q.where(AuditLog.timestamp >= datetime.fromisoformat(date_from))
|
||||||
|
if date_to:
|
||||||
|
q = q.where(AuditLog.timestamp <= datetime.fromisoformat(date_to))
|
||||||
|
|
||||||
|
q = q.order_by(AuditLog.timestamp.desc()).limit(10000)
|
||||||
|
result = await db.execute(q)
|
||||||
|
entries = result.scalars().all()
|
||||||
|
|
||||||
|
if format == "json":
|
||||||
|
content = json.dumps([_audit_to_dict(a) for a in entries], indent=2, default=str)
|
||||||
|
return StreamingResponse(
|
||||||
|
io.BytesIO(content.encode("utf-8")),
|
||||||
|
media_type="application/json",
|
||||||
|
headers={"Content-Disposition": "attachment; filename=audit_log_export.json"},
|
||||||
|
)
|
||||||
|
|
||||||
|
# CSV export
|
||||||
|
output = io.StringIO()
|
||||||
|
writer = csv.writer(output)
|
||||||
|
writer.writerow(["id", "timestamp", "user_id", "action", "entity_type", "entity_id", "tenant_id"])
|
||||||
|
for a in entries:
|
||||||
|
writer.writerow([
|
||||||
|
str(a.id), a.timestamp.isoformat() if a.timestamp else "",
|
||||||
|
str(a.user_id) if a.user_id else "",
|
||||||
|
a.action, a.entity_type,
|
||||||
|
str(a.entity_id) if a.entity_id else "",
|
||||||
|
str(a.tenant_id),
|
||||||
|
])
|
||||||
|
content = output.getvalue()
|
||||||
|
return StreamingResponse(
|
||||||
|
io.BytesIO(content.encode("utf-8")),
|
||||||
|
media_type="text/csv",
|
||||||
|
headers={"Content-Disposition": "attachment; filename=audit_log_export.csv"},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.delete("/retention")
|
||||||
|
async def audit_retention_cleanup(
|
||||||
|
retention_days: int = Query(365, ge=1, le=3650, description="Delete entries older than N days"),
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
|
current_user: dict = Depends(require_permission("audit:read")),
|
||||||
|
):
|
||||||
|
"""Delete audit log entries older than retention_days. Admin only.
|
||||||
|
|
||||||
|
Default retention: 365 days.
|
||||||
|
"""
|
||||||
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
||||||
|
cutoff = datetime.utcnow() - timedelta(days=retention_days)
|
||||||
|
|
||||||
|
q = delete(AuditLog).where(
|
||||||
|
AuditLog.tenant_id == tenant_id,
|
||||||
|
AuditLog.timestamp < cutoff,
|
||||||
|
)
|
||||||
|
result = await db.execute(q)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
return {"deleted": result.rowcount, "retention_days": retention_days, "cutoff": cutoff.isoformat()}
|
||||||
|
|||||||
Reference in New Issue
Block a user