From 2a173c9909d9ce3ceac92192a0cfb97f92dad1f9 Mon Sep 17 00:00:00 2001 From: Agent Zero Date: Thu, 20 Aug 2026 13:50:38 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20punkt=207=20(audit=20log)=20=E2=80=94?= =?UTF-8?q?=20export=20route=20(CSV/JSON),=20retention=20cleanup=20ARQ=20c?= =?UTF-8?q?ron=20job=20(daily=2003:00,=20365=20days=20default)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/worker.py | 48 ++++++++++++++++++++++++ app/routes/audit.py | 91 +++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 136 insertions(+), 3 deletions(-) diff --git a/app/core/worker.py b/app/core/worker.py index 041b92c..dcf3209 100644 --- a/app/core/worker.py +++ b/app/core/worker.py @@ -342,6 +342,49 @@ async def cleanup_outbox_job(ctx: dict[str, Any]) -> None: 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: """ARQ worker settings.""" functions = get_all_jobs() @@ -371,6 +414,11 @@ class WorkerSettings: _wrap_cron_with_lock("cleanup_outbox", cleanup_outbox_job, ttl_seconds=300), 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) cron( _wrap_cron_with_lock("run_backup", get_job("run_backup"), ttl_seconds=600), diff --git a/app/routes/audit.py b/app/routes/audit.py index 8f4f83f..7dd7c03 100644 --- a/app/routes/audit.py +++ b/app/routes/audit.py @@ -1,12 +1,16 @@ -"""Audit log routes — admin-only, paginated, filterable.""" +"""Audit log routes — admin-only, paginated, filterable, export, retention.""" from __future__ import annotations +import csv +import io +import json import uuid -from datetime import datetime +from datetime import datetime, timedelta 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 app.core.db import get_db @@ -92,3 +96,84 @@ async def list_audit_logs( "page_size": page_size, "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()}