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