Netdata.cloud bot for Zulip
anil.recoil.org
1"""Pydantic models for Netdata Cloud webhook payloads."""
2
3from datetime import datetime
4from enum import Enum
5from typing import Optional, Union
6
7from pydantic import BaseModel, Field, field_validator, ConfigDict
8
9
10class AlertSeverity(str, Enum):
11 """Alert severity levels."""
12 WARNING = "warning"
13 CRITICAL = "critical"
14 CLEAR = "clear"
15
16
17class ReachabilitySeverity(str, Enum):
18 """Reachability severity levels."""
19 INFO = "info"
20 CRITICAL = "critical"
21
22
23class ReachabilityStatus(BaseModel):
24 """Reachability status information."""
25 reachable: bool
26 text: str # "reachable" or "unreachable"
27
28
29class AlertNotification(BaseModel):
30 """Alert notification payload from Netdata Cloud."""
31 message: str
32 alert: str
33 info: str
34 chart: str
35 context: str
36 space: str
37 severity: AlertSeverity
38 date: datetime
39 alert_url: str
40 # Additional fields from full schema
41 Rooms: Optional[dict] = None
42 family: Optional[str] = None
43 class_: Optional[str] = Field(None, alias="class") # 'class' is a Python keyword
44 duration: Optional[str] = None
45 additional_active_critical_alerts: Optional[int] = None
46 additional_active_warning_alerts: Optional[int] = None
47
48 @field_validator('date', mode='before')
49 @classmethod
50 def parse_date(cls, v):
51 if isinstance(v, str):
52 return datetime.fromisoformat(v.replace('Z', '+00:00'))
53 return v
54
55
56class ReachabilityNotification(BaseModel):
57 """Reachability notification payload from Netdata Cloud."""
58 message: str
59 url: str
60 host: str
61 severity: ReachabilitySeverity
62 status: ReachabilityStatus
63
64
65class TestNotification(BaseModel):
66 """Test notification payload from Netdata Cloud."""
67 message: str
68
69
70class WebhookPayload(BaseModel):
71 """Union type for webhook payloads."""
72
73 @classmethod
74 def parse(cls, data: dict) -> Union[AlertNotification, ReachabilityNotification, TestNotification]:
75 """Parse webhook payload and determine notification type."""
76 if 'alert' in data and 'chart' in data:
77 return AlertNotification(**data)
78 elif 'status' in data and 'host' in data:
79 return ReachabilityNotification(**data)
80 elif len(data) == 1 and 'message' in data:
81 # Test notification - only has a message field
82 return TestNotification(**data)
83 else:
84 raise ValueError(f"Unknown notification type: {data}")
85
86
87class ZulipConfig(BaseModel):
88 """Zulip bot configuration."""
89 site: Optional[str] = None
90 email: Optional[str] = None
91 api_key: Optional[str] = None
92 stream: str = "netdata-alerts"
93
94 model_config = ConfigDict(env_prefix="ZULIP_")
95
96
97class ServerConfig(BaseModel):
98 """Server configuration."""
99 host: str = "0.0.0.0"
100 port: int = 8080 # Default HTTP port
101 challenge_secret: Optional[str] = None # Netdata webhook challenge secret
102
103 model_config = ConfigDict(env_prefix="SERVER_")