Spaces:
Sleeping
Sleeping
File size: 2,520 Bytes
5e3877e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
"""
Audit Service - Handle security and activity logging.
"""
import logging
from datetime import datetime
from typing import Optional, Dict, Any
from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
from core.models import AuditLog
logger = logging.getLogger(__name__)
class AuditService:
"""
Service for creating audit logs.
"""
@staticmethod
async def log_event(
db: AsyncSession,
action: str,
status: str,
user_id: Optional[int] = None,
client_user_id: Optional[str] = None,
details: Optional[Dict[str, Any]] = None,
request: Optional[Request] = None,
error_message: Optional[str] = None,
log_type: str = "client"
) -> AuditLog:
"""
Create an audit log entry.
Args:
db: Database session
action: Event action (e.g., "login", "logout")
status: Outcome (e.g., "success", "failure")
user_id: ID of the authenticated user (if any)
client_user_id: Client-side ID (if any)
details: Additional metadata
request: FastAPI request object (for IP/UA)
error_message: Error description if failed
log_type: "client" or "server"
Returns:
The created AuditLog instance
"""
ip_address = None
user_agent = None
refer_url = None
if request:
ip_address = request.client.host if request.client else None
user_agent = request.headers.get("user-agent")
refer_url = request.headers.get("referer")
# Create log entry
audit_log = AuditLog(
log_type=log_type,
user_id=user_id,
client_user_id=client_user_id,
action=action,
details=details or {},
ip_address=ip_address,
user_agent=user_agent,
refer_url=refer_url,
status=status,
error_message=error_message,
timestamp=datetime.utcnow()
)
db.add(audit_log)
# We don't commit here to allow the caller to group with other transactions
# or commit explicitly.
logger.info(f"Audit Log: [{status}] {action} - User: {user_id or 'Anon'} - IP: {ip_address}")
return audit_log
# Global instance not strictly needed as it's a static method helper,
# but keeping pattern consistent if we add state later.
audit_service = AuditService()
|