43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
"""Audit logging service."""
|
|
from datetime import datetime
|
|
from typing import Optional, Dict, Any
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.audit import AuditLog
|
|
|
|
class AuditService:
|
|
@classmethod
|
|
async def log(
|
|
cls,
|
|
db: AsyncSession,
|
|
action: str,
|
|
user_id: Optional[int] = None,
|
|
organization_id: Optional[int] = None,
|
|
resource_type: Optional[str] = None,
|
|
resource_id: Optional[int] = None,
|
|
old_values: Optional[Dict[str, Any]] = None,
|
|
new_values: Optional[Dict[str, Any]] = None,
|
|
description: Optional[str] = None,
|
|
ip_address: Optional[str] = None,
|
|
user_agent: Optional[str] = None,
|
|
success: str = "success",
|
|
error_message: Optional[str] = None
|
|
):
|
|
"""Create an audit log entry."""
|
|
log = AuditLog(
|
|
action=action,
|
|
user_id=user_id,
|
|
organization_id=organization_id,
|
|
resource_type=resource_type,
|
|
resource_id=resource_id,
|
|
old_values=old_values,
|
|
new_values=new_values,
|
|
description=description,
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
success=success,
|
|
error_message=error_message
|
|
)
|
|
db.add(log)
|
|
await db.flush()
|
|
return log
|