mirrors/lego-monitoring
1
0
Fork 0
mirror of https://forgejo.altau.su/lego/lego-monitoring.git synced 2026-03-09 20:31:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

stop if alert cannot be sent

Browse source
This commit is contained in:
Alex Tau 2026-01-01 16:49:13 +03:00
parent 191839d30f
commit 6c8ae03b6a
1 changed files with 4 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

8
src/lego_monitoring/alerting/sender.py
View file

@ -40,14 +40,14 @@ def format_message(alert: Alert, note: str) -> str:
async def send_alert(alert: Alert, note: str = "") -> Success[None] | Failure[tenacity.RetryError]: async def send_alert(alert: Alert, note: str = "") -> Success[None] | Failure[tenacity.RetryError]:
await log_errors_async(_send_alert(alert, note)) return await log_errors_async(_send_alert(alert, note))
async def send_healthchecks_status(alert: Alert) -> Success[None] | Failure[tenacity.RetryError]: async def send_healthchecks_status(alert: Alert) -> Success[None] | Failure[tenacity.RetryError]:
await log_errors_async(_send_healthchecks_status(alert)) return await log_errors_async(_send_healthchecks_status(alert))
@tenacity.retry(wait=tenacity.wait_random_exponential(multiplier=1, max=60)) @tenacity.retry(wait=tenacity.wait_random_exponential(multiplier=1, max=60), stop=tenacity.stop_after_attempt(3))
async def _send_alert(alert: Alert, note: str = "") -> None: async def _send_alert(alert: Alert, note: str = "") -> None:
logging.debug(f"Sending {alert.alert_type} alert to Telegram") logging.debug(f"Sending {alert.alert_type} alert to Telegram")
tg_client = cvars.tg_client.get() tg_client = cvars.tg_client.get()
@ -57,7 +57,7 @@ async def _send_alert(alert: Alert, note: str = "") -> None:
await tg_client.send_message(entity=room_id, message=message) await tg_client.send_message(entity=room_id, message=message)
@tenacity.retry(wait=tenacity.wait_random_exponential(multiplier=1, max=60)) @tenacity.retry(wait=tenacity.wait_random_exponential(multiplier=1, max=60), stop=tenacity.stop_after_attempt(3))
async def _send_healthchecks_status(alert: Alert) -> None: async def _send_healthchecks_status(alert: Alert) -> None:
def get_pinging_key(keys: dict[str, str]): def get_pinging_key(keys: dict[str, str]):
if alert.healthchecks_slug in keys: if alert.healthchecks_slug in keys: