from sqlalchemy.orm import Session from typing import Optional from app.models.audit import AuditLog, AuditAction from app.models.user import User from app.models.tenant import Organization import json class AuditService: def __init__(self, db: Session): self.db = db def log_action( self, organization_id: int, action: AuditAction, resource_type: str, user_id: Optional[int] = None, resource_id: Optional[str] = None, details: Optional[dict] = None, ip_address: Optional[str] = None, user_agent: Optional[str] = None ): audit_log = AuditLog( organization_id=organization_id, user_id=user_id, action=action, resource_type=resource_type, resource_id=resource_id, details=json.dumps(details) if details else None, ip_address=ip_address, user_agent=user_agent ) self.db.add(audit_log) self.db.commit() return audit_log def log_user_activity( self, user: User, action: AuditAction, resource_type: str, resource_id: Optional[str] = None, details: Optional[dict] = None, ip_address: Optional[str] = None, user_agent: Optional[str] = None ): return self.log_action( organization_id=user.organization_id, user_id=user.id, action=action, resource_type=resource_type, resource_id=resource_id, details=details, ip_address=ip_address, user_agent=user_agent ) def log_organization_activity( self, organization: Organization, action: AuditAction, resource_type: str, user_id: Optional[int] = None, resource_id: Optional[str] = None, details: Optional[dict] = None, ip_address: Optional[str] = None, user_agent: Optional[str] = None ): return self.log_action( organization_id=organization.id, user_id=user_id, action=action, resource_type=resource_type, resource_id=resource_id, details=details, ip_address=ip_address, user_agent=user_agent )