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(
35            "post", config.WEBHOOK_ADDRESS, json=data, headers=headers
36        )
37        if r.ok:
38            return True
39        logger.error(r)
40    except Exception as err:
41        logger.error(err)
42    return False
43
44
45async def send_notifications():
46    if not queue:
47        return
48
49    notifications_to_send = list()
50    try:
51        while notification := queue.popleft():
52            if notification.tries > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
53                continue
54            if notification.send_at > dt.utcnow().timestamp():
55                queue.append(
56                    notification
57                )  # add it to the queue again for the next check
58                continue
59            notifications_to_send.append(notification)
60    except IndexError:  # if the queue is empty
61        pass
62
63    if not notifications_to_send:
64        return
65    if not await send([jsonable_encoder(notif) for notif in notifications_to_send]):
66        for notification in notifications_to_send:
67            if (notification.tries + 1) > config.NUMBER_OF_RECURRENT_NOTIFICATIONS:
68                continue
69            notification.tries += 1
70            notification.send_at = (  # schedule notification for n seconds later
71                dt.utcnow() + td(seconds=config.RECURRENT_NOTIFICATIONS_TIMEOUT)
72            ).timestamp()
73            queue.append(notification)
74
75
76def delete_expired_reminders() -> None:
77    with GetDB() as db:
78        db.query(NotificationReminder).filter(
79            NotificationReminder.expires_at < dt.utcnow()
80        ).delete()
81        db.commit()