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