Create a new alert.
(self, alert_type: str, severity: str, message: str)
| 193 | ) |
| 194 | |
| 195 | def _create_alert(self, alert_type: str, severity: str, message: str): |
| 196 | """Create a new alert.""" |
| 197 | alert = Alert( |
| 198 | id=f"{alert_type}_{int(time.time())}", |
| 199 | type=alert_type, |
| 200 | severity=severity, |
| 201 | message=message, |
| 202 | timestamp=datetime.now().isoformat(), |
| 203 | ) |
| 204 | |
| 205 | # Check if similar alert already exists |
| 206 | existing_alert = next( |
| 207 | (a for a in self.alerts if a.type == alert_type and not a.resolved), None |
| 208 | ) |
| 209 | if not existing_alert: |
| 210 | self.alerts.append(alert) |
| 211 | self._broadcast_alert(alert) |
| 212 | |
| 213 | def _broadcast_metrics(self, metrics: PerformanceMetrics): |
| 214 | """Broadcast metrics to all websocket connections.""" |
no test coverage detected