Michael Chihlas
7882b4723b
Per spec §5.6.1, audit rows are written at session terminal events
(resolve, escalate, escalate_without_walk). log_audit gains an optional
acting_as parameter that propagates the session's acting_as tag
('l1_coverage' for engineer coverers, null for native L1 users).
Final code review flagged this as Important — column existed but was
never populated. Four new integration tests cover all three paths.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
"""Centralized audit logging for admin and destructive actions."""
|
|
from uuid import UUID
|
|
from typing import Optional
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.audit_log import AuditLog
|
|
|
|
|
|
async def log_audit(
|
|
db: AsyncSession,
|
|
user_id: UUID,
|
|
action: str,
|
|
resource_type: str,
|
|
resource_id: Optional[UUID] = None,
|
|
details: Optional[dict] = None,
|
|
account_id: Optional[UUID] = None,
|
|
acting_as: Optional[str] = None,
|
|
) -> None:
|
|
"""Record an audit log entry. Does not commit — caller's commit picks it up.
|
|
|
|
acting_as: optional tag from the session (e.g. 'l1_coverage' for engineers
|
|
on the L1 surface, None for native l1_tech users).
|
|
"""
|
|
if account_id is None:
|
|
# Derive from the acting user's account as a fallback (one extra query).
|
|
from sqlalchemy import select
|
|
from app.models.user import User
|
|
result = await db.execute(
|
|
select(User.account_id).where(User.id == user_id)
|
|
)
|
|
account_id = result.scalar_one()
|
|
|
|
entry = AuditLog(
|
|
user_id=user_id,
|
|
account_id=account_id,
|
|
action=action,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
details=details,
|
|
acting_as=acting_as,
|
|
)
|
|
db.add(entry)
|