1import logging
 2from datetime import datetime as dt
 3from datetime import timedelta as td
 4from typing import Any, Dict, List
 5
 6import aiohttp
 7from fastapi.encoders import jsonable_encoder
 8from requests import Session
 9
10import config
11from app.db.models import NotificationReminder
12from app.utils.notification import queue
13from app.db import GetDB
14
15logger = logging.getLogger(__name__)
16
17session = Session()
18
19headers = {"x-webhook-secret": config.WEBHOOK_SECRET} if config.WEBHOOK_SECRET else None
20
21
22async def send(data: List[Dict[Any, Any]]) -> bool:
23    """Send the notification to the webhook address provided by config.WEBHOOK_ADDRESS
24
25    Args:
26        data (List[Dict[Any, Any]]): list of json encoded notifications
27
28    Returns:
29        bool: returns True if an ok response received
30    """
31    try:
32        logger.debug(f"Sending {len(data)} webhook updates to {config.WEBHOOK_ADDRESS}")
33
34        r = await aiohttp.request("post", config.WEBHOOK_ADDRESS, json=data, headers=headers)
35        if r.ok:
36            return True
37        logger.error(r)
38    except Exception as err:
39        logger.error(err)
40    return False
41
42
43async def send_notifications():
44    if not queue:
45        return
46
47    notifications_to_send = list()
48    try:
49        while (notification := queue.popleft()):
50            if (notification.tries > config.NUMBER_OF_RECURRENT_NOTIFICATIONS):
51                continue
52            if notification.send_at > dt.utcnow().timestamp():
53                queue.append(notification)  # add it to the queue again for the next check
54                continue
55            notifications_to_send.append(notification)
56    except IndexError:  # if the queue is empty
57        pass
58
59    if not notifications_to_send:
60        return
61    if not await send([jsonable_encoder(notif) for notif in notifications_to_send]):
62        for notification in notifications_to_send:
63            if (notification.tries + 1) > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
64                continue
65            notification.tries += 1
66            notification.send_at = (  # schedule notification for n seconds later
67                dt.utcnow() + td(seconds=config.RECURRENT_NOTIFICATIONS_TIMEOUT)).timestamp()
68            queue.append(notification)
69
70
71def delete_expired_reminders() -> None:
72    with GetDB() as db:
73        db.query(NotificationReminder).filter(NotificationReminder.expires_at < dt.utcnow()).delete()
74        db.commit()
75