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