Spaces:
Sleeping
Sleeping
| """ | |
| 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. | |
| """ | |
| 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() | |