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