58 lines
2.2 KiB
Python
58 lines
2.2 KiB
Python
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Enum as SQLEnum
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from app.db.session import Base
|
|
|
|
|
|
class AlertSeverity(str, Enum):
|
|
CRITICAL = "critical"
|
|
WARNING = "warning"
|
|
INFO = "info"
|
|
|
|
|
|
class AlertStatus(str, Enum):
|
|
ACTIVE = "active"
|
|
ACKNOWLEDGED = "acknowledged"
|
|
RESOLVED = "resolved"
|
|
|
|
|
|
class Alert(Base):
|
|
__tablename__ = "alerts"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
severity = Column(SQLEnum(AlertSeverity), default=AlertSeverity.WARNING)
|
|
status = Column(SQLEnum(AlertStatus), default=AlertStatus.ACTIVE)
|
|
datasource_id = Column(Integer, nullable=True, index=True)
|
|
datasource_name = Column(String(255), nullable=True)
|
|
message = Column(Text)
|
|
alert_metadata = Column(Text, nullable=True)
|
|
acknowledged_by = Column(Integer, nullable=True)
|
|
resolved_by = Column(Integer, nullable=True)
|
|
resolution_notes = Column(Text, nullable=True)
|
|
created_at = Column(DateTime, default=datetime.utcnow)
|
|
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
|
|
acknowledged_at = Column(DateTime, nullable=True)
|
|
resolved_at = Column(DateTime, nullable=True)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"id": self.id,
|
|
"severity": self.severity.value if self.severity else None,
|
|
"status": self.status.value if self.status else None,
|
|
"datasource_id": self.datasource_id,
|
|
"datasource_name": self.datasource_name,
|
|
"message": self.message,
|
|
"alert_metadata": self.alert_metadata,
|
|
"acknowledged_by": self.acknowledged_by,
|
|
"resolved_by": self.resolved_by,
|
|
"resolution_notes": self.resolution_notes,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None,
|
|
"acknowledged_at": self.acknowledged_at.isoformat() if self.acknowledged_at else None,
|
|
"resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
|
|
}
|