Files
orion/app/modules/monitoring/services/audit_provider.py
Samir Boulahtit f20266167d
Some checks failed
CI / ruff (push) Failing after 7s
CI / pytest (push) Failing after 1s
CI / architecture (push) Failing after 9s
CI / dependency-scanning (push) Successful in 27s
CI / audit (push) Successful in 8s
CI / docs (push) Has been skipped
fix(lint): auto-fix ruff violations and tune lint rules
- Auto-fixed 4,496 lint issues (import sorting, modern syntax, etc.)
- Added ignore rules for patterns intentional in this codebase:
  E402 (late imports), E712 (SQLAlchemy filters), B904 (raise from),
  SIM108/SIM105/SIM117 (readability preferences)
- Added per-file ignores for tests and scripts
- Excluded broken scripts/rename_terminology.py (has curly quotes)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-12 23:10:42 +01:00

79 lines
2.1 KiB
Python

# 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 typing import TYPE_CHECKING
from sqlalchemy.orm import Session
from app.modules.contracts.audit import AuditEvent
from app.modules.tenancy.models import AdminAuditLog
if TYPE_CHECKING:
pass
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:
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 Exception 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"]