mirror of
https://forgejo.altau.su/lego/lego-monitoring.git
synced 2026-03-10 04:41:10 +00:00
53 lines
2 KiB
Python
53 lines
2 KiB
Python
import asyncio
|
|
import datetime
|
|
import logging
|
|
from typing import Callable, Coroutine
|
|
|
|
from alerting import alerts
|
|
|
|
|
|
async def _call_check(check: Callable | Coroutine, *args, **kwargs) -> list[alerts.Alert]:
|
|
if isinstance(check, Callable):
|
|
result = check(*args, **kwargs)
|
|
if isinstance(result, Coroutine):
|
|
result = await result
|
|
elif isinstance(check, Coroutine):
|
|
result = await check
|
|
else:
|
|
raise TypeError(f"check is {type(check)}, neither function nor coroutine")
|
|
return result
|
|
|
|
|
|
async def interval_checker(check: Callable | Coroutine, interval: datetime.timedelta, *args, **kwargs):
|
|
interval_secs = interval.total_seconds()
|
|
while True:
|
|
logging.info(f"Calling {check.__name__}")
|
|
result = await _call_check(check, *args, **kwargs)
|
|
logging.info(f"Got {len(result)} alerts")
|
|
for alert in result:
|
|
await alerts.send_alert(alert)
|
|
await asyncio.sleep(interval_secs)
|
|
|
|
|
|
async def scheduled_checker(
|
|
check: Callable | Coroutine, period: datetime.timedelta, when: datetime.time, *args, **kwargs
|
|
):
|
|
match period:
|
|
case datetime.timedelta(days=1):
|
|
while True:
|
|
now = datetime.datetime.now()
|
|
next_datetime = datetime.datetime.combine(datetime.date.today(), when)
|
|
if next_datetime < now:
|
|
next_datetime += datetime.timedelta(days=1)
|
|
logging.info(f"Scheduled to call {check.__name__} at {next_datetime.isoformat()}")
|
|
await asyncio.sleep(
|
|
(next_datetime - now).total_seconds()
|
|
) # might be negative at this point, asyncio doesn't care
|
|
|
|
logging.info(f"Calling {check.__name__}")
|
|
result = await _call_check(check, *args, **kwargs)
|
|
logging.info(f"Got {len(result)} alerts")
|
|
for alert in result:
|
|
await alerts.send_alert(alert)
|
|
case _:
|
|
raise NotImplementedError
|