Spaces:
Runtime error
Runtime error
| """ | |
| 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"<Alert {self.id}: {self.title}>" | |
| 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"<Incident {self.id}: {self.title}>" | |
| 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"<IncidentEvent {self.id}: {self.event_type}>" | |
| 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"<IncidentAnalysis {self.id}: {self.analysis_type}>" | |
| 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"<IncidentRecommendation {self.id}: {self.recommendation}>" | |
| 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"<NotificationLog {self.id}: {self.channel} to {self.recipient}>" | |
| 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"<AuditLog {self.id}: {self.action}>" | |