""" Database models for alerts, incidents, and analyses """ from sqlalchemy import Column, String, Integer, Float, DateTime, Boolean, Enum, ForeignKey, JSON, Text from sqlalchemy.orm import relationship from sqlalchemy.dialects.postgresql import UUID import uuid from datetime import datetime import enum from src.database import Base class AlertSeverity(str, enum.Enum): """Alert severity levels""" CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" INFO = "INFO" class AlertStatus(str, enum.Enum): """Alert status""" NEW = "NEW" ACKNOWLEDGED = "ACKNOWLEDGED" RESOLVED = "RESOLVED" SUPPRESSED = "SUPPRESSED" class IncidentSeverity(str, enum.Enum): """Incident severity levels""" CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" LOW = "LOW" class IncidentStatus(str, enum.Enum): """Incident status""" OPEN = "OPEN" IN_PROGRESS = "IN_PROGRESS" RESOLVED = "RESOLVED" CLOSED = "CLOSED" class IncidentEventType(str, enum.Enum): """Types of incident events""" CREATED = "CREATED" UPDATED = "UPDATED" ANALYZED = "ANALYZED" RECOMMENDED = "RECOMMENDED" ACKNOWLEDGED = "ACKNOWLEDGED" RESOLVED = "RESOLVED" ESCALATED = "ESCALATED" COMMENTED = "COMMENTED" class AnalysisType(str, enum.Enum): """Types of LLM analysis""" CLASSIFICATION = "CLASSIFICATION" ROOT_CAUSE = "ROOT_CAUSE" RECOMMENDATION = "RECOMMENDATION" class Alert(Base): """Alert model""" __tablename__ = "alerts" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) source = Column(String(100), nullable=False, index=True) source_id = Column(String(255), nullable=False) title = Column(String(255), nullable=False) description = Column(Text) severity = Column(Enum(AlertSeverity), nullable=False, index=True) status = Column(Enum(AlertStatus), default=AlertStatus.NEW, nullable=False, index=True) # Categorization category = Column(String(100)) affected_service = Column(JSON) # List of services affected_component = Column(String(255)) tags = Column(JSON, default=[]) # Fingerprint for deduplication fingerprint = Column(String(255), nullable=False, unique=True, index=True) # Metrics and metadata metrics = Column(JSON) metadata = Column(JSON) remediation_hints = Column(JSON) # Relationships incident_id = Column(UUID(as_uuid=True), ForeignKey("incidents.id")) incident = relationship("Incident", back_populates="alerts") # Timestamps created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) resolved_at = Column(DateTime) def __repr__(self): return f"" class Incident(Base): """Incident model""" __tablename__ = "incidents" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) title = Column(String(255), nullable=False) description = Column(Text) # Classification severity = Column(Enum(IncidentSeverity), nullable=False, index=True) status = Column(Enum(IncidentStatus), default=IncidentStatus.OPEN, nullable=False, index=True) category = Column(String(100)) # Impact impact_summary = Column(Text) affected_services = Column(JSON, default=[]) estimated_users_impacted = Column(Integer) business_impact = Column(String(50)) # High, Medium, Low # Root cause and analysis root_cause = Column(Text) root_cause_confidence = Column(Float) # Resolution assigned_to = Column(String(255)) resolved_at = Column(DateTime) resolution_time_minutes = Column(Integer) # Relationships alerts = relationship("Alert", back_populates="incident") events = relationship("IncidentEvent", back_populates="incident", cascade="all, delete-orphan") analyses = relationship("IncidentAnalysis", back_populates="incident", cascade="all, delete-orphan") recommendations = relationship("IncidentRecommendation", back_populates="incident", cascade="all, delete-orphan") # Timestamps created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) def __repr__(self): return f"" class IncidentEvent(Base): """Timeline event for an incident""" __tablename__ = "incident_events" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) incident_id = Column(UUID(as_uuid=True), ForeignKey("incidents.id"), nullable=False) incident = relationship("Incident", back_populates="events") event_type = Column(Enum(IncidentEventType), nullable=False, index=True) message = Column(Text, nullable=False) actor = Column(String(255)) # SYSTEM or user email details = Column(JSON) created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) def __repr__(self): return f"" class IncidentAnalysis(Base): """LLM analysis results for an incident""" __tablename__ = "llm_analyses" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) incident_id = Column(UUID(as_uuid=True), ForeignKey("incidents.id"), nullable=False) incident = relationship("Incident", back_populates="analyses") analysis_type = Column(Enum(AnalysisType), nullable=False, index=True) model_used = Column(String(100), nullable=False) # Prompt and response prompt = Column(Text) response = Column(Text) response_json = Column(JSON) # Confidence confidence_score = Column(Float) # Metadata tokens_used = Column(Integer) latency_ms = Column(Integer) created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) def __repr__(self): return f"" class IncidentRecommendation(Base): """Recommended actions for an incident""" __tablename__ = "incident_recommendations" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) incident_id = Column(UUID(as_uuid=True), ForeignKey("incidents.id"), nullable=False) incident = relationship("Incident", back_populates="recommendations") priority = Column(String(50)) # CRITICAL, HIGH, MEDIUM, LOW phase = Column(String(50)) # immediate, investigation, remediation, prevention recommendation = Column(String(255), nullable=False) description = Column(Text) estimated_time = Column(String(100)) risk_level = Column(String(50)) # Critical, High, Medium, Low success_criteria = Column(Text) # Status status = Column(String(50), default="PENDING") # PENDING, IN_PROGRESS, COMPLETED, SKIPPED executed_at = Column(DateTime) executed_by = Column(String(255)) created_at = Column(DateTime, default=datetime.utcnow, nullable=False) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) def __repr__(self): return f"" class NotificationLog(Base): """Log of sent notifications""" __tablename__ = "notification_logs" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) incident_id = Column(UUID(as_uuid=True), ForeignKey("incidents.id")) channel = Column(String(50), nullable=False) # email, slack, webhook, etc recipient = Column(String(255), nullable=False) subject = Column(String(255)) body = Column(Text) status = Column(String(50), default="SENT") # SENT, FAILED, BOUNCED error_message = Column(Text) created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) def __repr__(self): return f"" class AuditLog(Base): """Audit log for user actions""" __tablename__ = "audit_logs" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) user = Column(String(255), nullable=False) action = Column(String(255), nullable=False) resource_type = Column(String(100)) resource_id = Column(String(255)) changes = Column(JSON) created_at = Column(DateTime, default=datetime.utcnow, nullable=False, index=True) def __repr__(self): return f""