63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
import logging
|
|
import uuid
|
|
from typing import Any, Dict, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.system.audit_log import AuditLog
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AuditLogService:
|
|
@staticmethod
|
|
def log(
|
|
db: Session,
|
|
module_name: str,
|
|
action_type: str,
|
|
description: str,
|
|
entity_id: Optional[str] = None,
|
|
entity_name: Optional[str] = None,
|
|
performed_by_id: Optional[str] = None,
|
|
performed_by_email: Optional[str] = None,
|
|
ip_address: Optional[str] = None,
|
|
old_values: Optional[Dict[str, Any]] = None,
|
|
new_values: Optional[Dict[str, Any]] = None,
|
|
commit: bool = True,
|
|
strict: bool = False,
|
|
) -> Optional[AuditLog]:
|
|
"""
|
|
Write an audit log entry.
|
|
|
|
If commit=True, commits the session. If commit=False, flushes to the active transaction.
|
|
If strict=True, exceptions are re-raised to the caller.
|
|
If strict=False, failures are swallowed and logged.
|
|
"""
|
|
try:
|
|
entry = AuditLog(
|
|
module_name=module_name,
|
|
action_type=action_type,
|
|
entity_id=entity_id,
|
|
entity_name=entity_name,
|
|
description=description,
|
|
performed_by_id=(
|
|
uuid.UUID(performed_by_id) if performed_by_id else None
|
|
),
|
|
performed_by_email=performed_by_email,
|
|
ip_address=ip_address,
|
|
old_values=old_values,
|
|
new_values=new_values,
|
|
)
|
|
db.add(entry)
|
|
if commit:
|
|
db.commit()
|
|
else:
|
|
db.flush()
|
|
return entry
|
|
except Exception as exc:
|
|
if strict:
|
|
raise
|
|
if commit:
|
|
db.rollback()
|
|
logger.error("AuditLogService.log failed: %s", exc, exc_info=True)
|
|
return None |