Coverage for chalicelib/core/alerts/alerts.py: 38%
146 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 12:56 +0000
1import json
2import logging
3import time
4from datetime import datetime
6from decouple import config
8import schemas
9from chalicelib.core import notifications, webhook
10from chalicelib.core.collaborations.collaboration_msteams import MSTeams
11from chalicelib.core.collaborations.collaboration_slack import Slack
12from chalicelib.utils import pg_client, helper, email_helper, smtp
13from chalicelib.utils.TimeUTC import TimeUTC
14from chalicelib.utils.log import sanitize
15from starlette import status
16from starlette.exceptions import HTTPException
18logger = logging.getLogger(__name__)
21def get(project_id, id):
22 try:
23 with pg_client.PostgresClient() as cur:
24 cur.execute(
25 cur.mogrify("""\
26 SELECT *
27 FROM public.alerts
28 WHERE alert_id =%(id)s AND project_id=%(project_id)s;""",
29 {"project_id": project_id, "id": id})
30 )
32 if cur.rowcount == 0: 32 ↛ 35line 32 didn't jump to line 35 because the condition on line 32 was always true
33 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Alert not found.")
35 a = helper.dict_to_camel_case(cur.fetchone())
37 return helper.custom_alert_to_front(__process_circular(a))
38 except Exception as e:
39 logger.error(f"Error fetching alert: {sanitize(str(e))}")
40 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Alert not found.")
43def get_all(project_id):
44 with pg_client.PostgresClient() as cur:
45 query = cur.mogrify("""\
46 SELECT alerts.*,
47 COALESCE(metrics.name || '.' || (COALESCE(metric_series.name, 'series ' || index)) || '.count',
48 query ->> 'left') AS series_name
49 FROM public.alerts
50 LEFT JOIN metric_series USING (series_id)
51 LEFT JOIN metrics USING (metric_id)
52 WHERE alerts.project_id =%(project_id)s
53 AND alerts.deleted_at ISNULL
54 ORDER BY alerts.created_at;""",
55 {"project_id": project_id})
56 cur.execute(query=query)
57 all = helper.list_to_camel_case(cur.fetchall())
58 for i in range(len(all)):
59 all[i] = helper.custom_alert_to_front(__process_circular(all[i]))
60 return all
63def __process_circular(alert):
64 if alert is None: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 return None
66 alert.pop("deletedAt")
67 alert["createdAt"] = TimeUTC.datetime_to_timestamp(alert["createdAt"])
68 return alert
71def create(project_id, data: schemas.AlertSchema):
72 data = data.model_dump()
73 data["query"] = json.dumps(data["query"])
74 data["options"] = json.dumps(data["options"])
76 with pg_client.PostgresClient() as cur:
77 cur.execute(
78 cur.mogrify("""\
79 INSERT INTO public.alerts(project_id, name, description, detection_method, query, options, series_id, change)
80 VALUES (%(project_id)s, %(name)s, %(description)s, %(detection_method)s, %(query)s, %(options)s::jsonb, %(series_id)s, %(change)s)
81 RETURNING *;""",
82 {"project_id": project_id, **data})
83 )
84 a = helper.dict_to_camel_case(cur.fetchone())
85 return {"data": helper.custom_alert_to_front(helper.dict_to_camel_case(__process_circular(a)))}
88def update(project_id: int, id: int, data: schemas.AlertSchema):
89 data = data.model_dump()
90 data["query"] = json.dumps(data["query"])
91 data["options"] = json.dumps(data["options"])
93 try:
94 with pg_client.PostgresClient() as cur:
95 query = cur.mogrify("""\
96 UPDATE public.alerts
97 SET name = %(name)s,
98 description = %(description)s,
99 active = TRUE,
100 detection_method = %(detection_method)s,
101 query = %(query)s,
102 options = %(options)s,
103 series_id = %(series_id)s,
104 change = %(change)s
105 WHERE alert_id =%(id)s AND project_id = %(project_id)s AND deleted_at ISNULL
106 RETURNING *;""",
107 {"project_id": project_id, "id": id, **data})
108 cur.execute(query=query)
110 if cur.rowcount == 0: 110 ↛ 113line 110 didn't jump to line 113 because the condition on line 110 was always true
111 raise ValueError(f"Alert with id {id} not found in project {project_id}.")
113 a = helper.dict_to_camel_case(cur.fetchone())
115 return {"data": helper.custom_alert_to_front(__process_circular(a))}
116 except Exception as e:
117 logger.error(f"Error updating alert: {sanitize(str(e))}")
118 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to update alert.")
121def process_notifications(data):
122 full = {}
123 for n in data:
124 if "message" in n["options"]:
125 webhook_data = {}
126 if "data" in n["options"]:
127 webhook_data = n["options"].pop("data")
128 for c in n["options"].pop("message"):
129 if c["type"] not in full:
130 full[c["type"]] = []
131 if c["type"] in ["slack", "msteams", "email"]:
132 full[c["type"]].append({
133 "notification": n,
134 "destination": c["value"]
135 })
136 elif c["type"] in ["webhook"]:
137 full[c["type"]].append({"data": webhook_data, "destination": c["value"]})
138 notifications.create(data)
139 BATCH_SIZE = 200
140 for t in full.keys():
141 for i in range(0, len(full[t]), BATCH_SIZE):
142 notifications_list = full[t][i:min(i + BATCH_SIZE, len(full[t]))]
143 if notifications_list is None or len(notifications_list) == 0:
144 break
146 if t == "slack":
147 try:
148 send_to_slack_batch(notifications_list=notifications_list)
149 except Exception as e:
150 logger.error("!!!Error while sending slack notifications batch")
151 logger.error(sanitize(str(e)))
152 elif t == "msteams":
153 try:
154 send_to_msteams_batch(notifications_list=notifications_list)
155 except Exception as e:
156 logger.error("!!!Error while sending msteams notifications batch")
157 logger.error(sanitize(str(e)))
158 elif t == "email":
159 try:
160 send_by_email_batch(notifications_list=notifications_list)
161 except Exception as e:
162 logger.error("!!!Error while sending email notifications batch")
163 logger.error(sanitize(str(e)))
164 elif t == "webhook":
165 try:
166 webhook.trigger_batch(data_list=notifications_list)
167 except Exception as e:
168 logger.error("!!!Error while sending webhook notifications batch")
169 logger.error(sanitize(str(e)))
172def send_by_email(notification, destination):
173 if notification is None:
174 return
175 email_helper.alert_email(recipients=destination,
176 subject=f'"{notification["title"]}" has been triggered',
177 data={
178 "message": f'"{notification["title"]}" {notification["description"]}',
179 "project_id": notification["options"]["projectId"]})
182def send_by_email_batch(notifications_list):
183 if not smtp.has_smtp():
184 logger.info("no SMTP configuration for email notifications")
185 if notifications_list is None or len(notifications_list) == 0:
186 logger.info("no email notifications")
187 return
188 for n in notifications_list:
189 send_by_email(notification=n.get("notification"), destination=n.get("destination"))
190 time.sleep(1)
193def send_to_slack_batch(notifications_list):
194 webhookId_map = {}
195 for n in notifications_list:
196 if n.get("destination") not in webhookId_map:
197 webhookId_map[n.get("destination")] = {"tenantId": n["notification"]["tenantId"], "batch": []}
198 webhookId_map[n.get("destination")]["batch"].append({"text": n["notification"]["description"] \
199 + f"\n<{config('SITE_URL')}{n['notification']['buttonUrl']}|{n['notification']['buttonText']}>",
200 "title": n["notification"]["title"],
201 "title_link": n["notification"]["buttonUrl"],
202 "ts": datetime.now().timestamp()})
203 for batch in webhookId_map.keys():
204 Slack.send_batch(tenant_id=webhookId_map[batch]["tenantId"], webhook_id=batch,
205 attachments=webhookId_map[batch]["batch"])
208def send_to_msteams_batch(notifications_list):
209 webhookId_map = {}
210 for n in notifications_list:
211 if n.get("destination") not in webhookId_map:
212 webhookId_map[n.get("destination")] = {"tenantId": n["notification"]["tenantId"], "batch": []}
214 link = f"{config('SITE_URL')}{n['notification']['buttonUrl']}"
215 # for MSTeams, the batch is the list of `sections`
216 webhookId_map[n.get("destination")]["batch"].append(
217 {
218 "activityTitle": n["notification"]["title"],
219 "activitySubtitle": f"On Project *{n['notification']['projectName']}*",
220 "facts": [
221 {
222 "name": "Target:",
223 "value": link
224 },
225 {
226 "name": "Description:",
227 "value": n["notification"]["description"]
228 }],
229 "markdown": True
230 }
231 )
232 for batch in webhookId_map.keys():
233 MSTeams.send_batch(tenant_id=webhookId_map[batch]["tenantId"], webhook_id=batch,
234 attachments=webhookId_map[batch]["batch"])
237def delete(project_id, alert_id):
238 with pg_client.PostgresClient() as cur:
239 cur.execute(
240 cur.mogrify(""" UPDATE public.alerts
241 SET deleted_at = timezone('utc'::text, now()),
242 active = FALSE
243 WHERE alert_id = %(alert_id)s AND project_id=%(project_id)s;""",
244 {"alert_id": alert_id, "project_id": project_id})
245 )
246 return {"data": {"state": "success"}}
249def get_predefined_values():
250 values = [e.value for e in schemas.AlertColumn]
251 values = [{"name": v, "value": v,
252 "unit": "count" if v.endswith(".count") else "ms",
253 "predefined": True,
254 "metricId": None,
255 "seriesId": None} for v in values if v != schemas.AlertColumn.CUSTOM]
256 return values