# app/modules/monitoring/services/audit_provider.py """ Audit provider implementation for the monitoring module. Provides database-backed audit logging using the AdminAuditLog model. This wraps the existing admin_audit_service functionality in the AuditProviderProtocol interface. """ import logging from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session from app.modules.contracts.audit import AuditEvent logger = logging.getLogger(__name__) class DatabaseAuditProvider: """ Database-backed audit provider. Logs admin actions to the AdminAuditLog table. This is the default audit backend for the platform. """ @property def audit_backend(self) -> str: return "database" def log_action(self, db: Session, event: AuditEvent) -> bool: """ Log an audit event to the database. Args: db: Database session event: The audit event to log Returns: True if logged successfully, False otherwise """ try: from app.modules.tenancy.models import AdminAuditLog audit_log = AdminAuditLog( admin_user_id=event.admin_user_id, action=event.action, target_type=event.target_type, target_id=str(event.target_id), details=event.details or {}, ip_address=event.ip_address, user_agent=event.user_agent, request_id=event.request_id, ) db.add(audit_log) db.flush() logger.debug( f"Admin action logged: {event.action} on {event.target_type}:" f"{event.target_id} by admin {event.admin_user_id}" ) return True except SQLAlchemyError as e: logger.error(f"Failed to log admin action: {str(e)}") # Don't raise exception - audit logging should not break operations return False # Singleton instance audit_provider = DatabaseAuditProvider() __all__ = ["DatabaseAuditProvider", "audit_provider"]