github/MIND
1
1
Fork 0
mirror of https://github.com/Casvt/MIND.git synced 2026-02-19 11:54:46 -05:00
Code Issues Packages Projects Releases Wiki Activity

Refactored reminders.py

Browse Source
This commit is contained in:
CasVT
2025-08-21 16:57:23 +02:00
parent 85a2f32dad
commit 61882e609a
4 changed files with 141 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
backend/features/reminder_handler.py
View File

@@ -6,10 +6,9 @@ from datetime import datetime
from typing import TYPE_CHECKING, Union
from backend.base.definitions import Constants, RepeatQuantity, SendResult
from backend.base.helpers import (Singleton, find_next_time,
send_apprise_notification, when_not_none)
from backend.base.helpers import Singleton, find_next_time, when_not_none
from backend.base.logging import LOGGER
from backend.implementations.notification_services import NotificationService
from backend.implementations.reminders import Reminder
from backend.internals.db_models import UserlessRemindersDB
from backend.internals.server import Server
@@ -40,14 +39,7 @@ class ReminderHandler(metaclass=Singleton):
user_id = self.reminder_db.reminder_id_to_user_id(
reminder.id
)
result = send_apprise_notification(
[
NotificationService(user_id, ns).get().url
for ns in reminder.notification_services
],
reminder.title,
reminder.text
)
result = Reminder(user_id, reminder.id).trigger_reminder()
self.thread = None
self.time = None

236
backend/implementations/reminders.py
View File

@@ -1,6 +1,5 @@
# -*- coding: utf-8 -*-
from dataclasses import asdict
from datetime import datetime
from typing import List, Union
@@ -12,16 +11,13 @@ from backend.base.definitions import (WEEKDAY_NUMBER, ReminderData,
from backend.base.helpers import (find_next_time, search_filter,
send_apprise_notification, when_not_none)
from backend.base.logging import LOGGER
from backend.features.reminder_handler import ReminderHandler
from backend.implementations.notification_services import NotificationService
from backend.internals.db_models import RemindersDB
REMINDER_HANDLER = ReminderHandler()
class Reminder:
def __init__(self, user_id: int, reminder_id: int) -> None:
"""Represent a reminder.
"""Create an instance.
Args:
user_id (int): The ID of the user.
@@ -29,11 +25,10 @@ class Reminder:
Raises:
ReminderNotFound: Reminder with given ID does not exist or is not
owned by user.
owned by user.
"""
self.user_id = user_id
self.id = reminder_id
self.reminder_db = RemindersDB(self.user_id)
if not self.reminder_db.exists(self.id):
@@ -48,72 +43,96 @@ class Reminder:
"""
return self.reminder_db.fetch(self.id)[0]
def trigger_reminder(self) -> SendResult:
"""Send the reminder.
Returns:
SendResult: The result of the sending process.
"""
LOGGER.info(f'Triggering reminder {self.id}')
reminder_data = self.get()
urls = [
NotificationService(self.user_id, ns_id).get().url
for ns_id in reminder_data.notification_services
]
return send_apprise_notification(
urls,
reminder_data.title,
reminder_data.text
)
def update(
self,
title: Union[None, str] = None,
time: Union[None, int] = None,
notification_services: Union[None, List[int]] = None,
text: Union[None, str] = None,
repeat_quantity: Union[None, RepeatQuantity] = None,
repeat_interval: Union[None, int] = None,
weekdays: Union[None, List[WEEKDAY_NUMBER]] = None,
cron_schedule: Union[None, str] = None,
color: Union[None, str] = None,
enabled: Union[None, bool] = None
title: Union[str, None] = None,
time: Union[int, None] = None,
notification_services: Union[List[int], None] = None,
text: Union[str, None] = None,
repeat_quantity: Union[RepeatQuantity, None] = None,
repeat_interval: Union[int, None] = None,
weekdays: Union[List[WEEKDAY_NUMBER], None] = None,
cron_schedule: Union[str, None] = None,
color: Union[str, None] = None,
enabled: Union[bool, None] = None
) -> ReminderData:
"""Edit the reminder.
Args:
title (Union[None, str]): The new title of the entry.
title (Union[str, None], optional): The new title of the reminder.
Defaults to None.
time (Union[None, int]): The new UTC epoch timestamp when the
reminder should be send.
time (Union[int, None], optional): The new UTC epoch timestamp when
the reminder should be send.
Defaults to None.
notification_services (Union[None, List[int]]): The new list
of id's of the notification services to use to send the reminder.
notification_services (Union[List[int], None], optional): The new
IDs of the notification services to use to send the reminder.
Defaults to None.
text (Union[None, str], optional): The new body of the reminder.
text (Union[str, None], optional): The new body of the reminder.
Defaults to None.
repeat_quantity (Union[None, RepeatQuantity], optional): The new
quantity of the repeat specified for the reminder.
repeat_quantity (Union[RepeatQuantity, None], optional): The new
quantity of the repeat_interval specified for the reminder.
Defaults to None.
repeat_interval (Union[None, int], optional): The new amount of
repeat_quantity, like "5" (hours).
repeat_interval (Union[int, None], optional): The new interval of
repeat_quantity, like "5" (hours).
Defaults to None.
weekdays (Union[None, List[WEEKDAY_NUMBER]], optional): The new
indexes of the days of the week that the reminder should run.
weekdays (Union[List[WEEKDAY_NUMBER], None], optional): The new
indexes of the days of the week that the reminder should run.
Defaults to None.
cron_schedule (Union[None, str], optional): The new cron schedule
that the reminder should run on.
cron_schedule (Union[str, None], optional): The new cron schedule
that the reminder should run on.
Defaults to None.
color (Union[None, str], optional): The new hex code of the color
of the reminder, which is shown in the web-ui.
color (Union[str, None], optional): The new hex code of the color
of the reminder, which is shown in the web-ui.
Defaults to None.
enabled (Union[None, bool], optional): Whether the reminder should
be enabled.
enabled (Union[bool, None], optional): Whether the reminder should
be enabled.
Defaults to None.
Note about args:
Either repeat_quantity and repeat_interval are given, weekdays is
given or neither, but not both.
given, cron_schedule is given, or none, but not a combination.
Raises:
NotificationServiceNotFound: One of the notification services was not found.
NotificationServiceNotFound: One of the notification services was
not found.
InvalidKeyValue: The value of one of the keys is not valid or
the "Note about args" is violated.
the "Note about args" is violated.
Returns:
ReminderData: The new reminder info.
"""
from backend.features.reminder_handler import ReminderHandler
LOGGER.info(
f'Updating notification service {self.id}: '
+ f'{title=}, {time=}, {notification_services=}, {text=}, '
@@ -137,9 +156,9 @@ class Reminder:
raise InvalidKeyValue('weekdays', weekdays)
repeated_reminder = (
(repeat_quantity is not None and repeat_interval is not None)
or weekdays is not None
or cron_schedule is not None
(repeat_quantity is not None and repeat_interval)
or weekdays
or cron_schedule
)
if time is not None:
@@ -149,12 +168,12 @@ class Reminder:
time = round(time)
if notification_services:
# Check if all notification services exist
# Check if all notification services exist. Creating an instance
# raises NotificationServiceNotFound if the ID is not valid.
for ns in notification_services:
NotificationService(self.user_id, ns)
# Get current data and update it with new values
data = asdict(self.get())
data = self.get().todict()
new_values = {
'title': title,
@@ -225,14 +244,16 @@ class Reminder:
data["enabled"]
)
REMINDER_HANDLER.find_next_reminder(next_time)
ReminderHandler().find_next_reminder(next_time)
return self.get()
def delete(self) -> None:
"Delete the reminder"
from backend.features.reminder_handler import ReminderHandler
LOGGER.info(f'Deleting reminder {self.id}')
self.reminder_db.delete(self.id)
REMINDER_HANDLER.find_next_reminder()
ReminderHandler().find_next_reminder()
return
@@ -247,18 +268,19 @@ class Reminders:
self.reminder_db = RemindersDB(self.user_id)
return
def fetchall(
def get_all(
self,
sort_by: SortingMethod = SortingMethod.TIME
) -> List[ReminderData]:
"""Get all reminders.
"""Get all reminders of the user.
Args:
sort_by (SortingMethod, optional): How to sort the result.
sort_by (SortingMethod, optional): How to sort the
reminders.
Defaults to SortingMethod.TIME.
Returns:
List[ReminderData]: The info of each reminder.
List[ReminderData]: The info about all reminders of the user.
"""
reminders = self.reminder_db.fetch()
reminders.sort(key=sort_by.value[0], reverse=sort_by.value[1])
@@ -273,34 +295,33 @@ class Reminders:
Args:
query (str): The term to search for.
sort_by (SortingMethod, optional): How to sort the result.
sort_by (SortingMethod, optional): How to sort the
reminders.
Defaults to SortingMethod.TIME.
Returns:
List[ReminderData]: All reminders that match. Similar output to
self.fetchall.
List[ReminderData]: The info about all reminders of the user that
match the search term.
"""
reminders = [
return [
r
for r in self.fetchall(sort_by)
for r in self.get_all(sort_by)
if search_filter(query, r)
]
return reminders
def fetchone(self, id: int) -> Reminder:
"""Get one reminder.
def get_one(self, reminder_id: int) -> Reminder:
"""Get a reminder instance based on the ID.
Args:
id (int): The ID of the reminder to fetch.
reminder_id (int): The ID of the reminder to fetch.
Raises:
ReminderNotFound: The reminder with the given ID does not exist
or is not owned by the user.
ReminderNotFound: The user does not own a reminder with the given ID.
Returns:
Reminder: A Reminder instance.
Reminder: Instance of Reminder.
"""
return Reminder(self.user_id, id)
return Reminder(self.user_id, reminder_id)
def add(
self,
@@ -308,66 +329,68 @@ class Reminders:
time: int,
notification_services: List[int],
text: str = '',
repeat_quantity: Union[None, RepeatQuantity] = None,
repeat_interval: Union[None, int] = None,
weekdays: Union[None, List[WEEKDAY_NUMBER]] = None,
cron_schedule: Union[None, str] = None,
color: Union[None, str] = None,
repeat_quantity: Union[RepeatQuantity, None] = None,
repeat_interval: Union[int, None] = None,
weekdays: Union[List[WEEKDAY_NUMBER], None] = None,
cron_schedule: Union[str, None] = None,
color: Union[str, None] = None,
enabled: bool = True
) -> Reminder:
"""Add a reminder.
Args:
title (str): The title of the entry.
title (str): The title of the reminder.
time (int): The UTC epoch timestamp the the reminder should be send.
notification_services (List[int]): The id's of the notification
services to use to send the reminder.
notification_services (List[int]): The IDs of the notification
services to use to send the reminder.
text (str, optional): The body of the reminder.
Defaults to ''.
repeat_quantity (Union[None, RepeatQuantity], optional): The quantity
of the repeat specified for the reminder.
repeat_quantity (Union[RepeatQuantity, None], optional): The
quantity of the repeat_interval specified for the reminder.
Defaults to None.
repeat_interval (Union[None, int], optional): The amount of
repeat_quantity, like "5" (hours).
repeat_interval (Union[int, None], optional): The interval of
repeat_quantity, like "5" (hours).
Defaults to None.
weekdays (Union[None, List[WEEKDAY_NUMBER]], optional): The indexes
of the days of the week that the reminder should run.
weekdays (Union[List[WEEKDAY_NUMBER], None], optional): The indexes
of the days of the week that the reminder should run.
Defaults to None.
cron_schedule (Union[None, str], optional): The cron schedule that
the reminder should run on.
cron_schedule (Union[str, None], optional): The cron schedule that
the reminder should run on.
Defaults to None.
color (Union[None, str], optional): The hex code of the color of the
reminder, which is shown in the web-ui.
color (Union[str, None], optional): The hex code of the color of the
reminder, which is shown in the web-ui.
Defaults to None.
enabled (Union[None, bool], optional): Whether the reminder should
be enabled.
enabled (Union[bool, None], optional): Whether the reminder should
be enabled.
Defaults to None.
Note about args:
Either repeat_quantity and repeat_interval are given,
weekdays is given, cron_schedule is given or none.
Either repeat_quantity and repeat_interval are given, weekdays is
given, cron_schedule is given, or none, but not a combination.
Raises:
NotificationServiceNotFound: One of the notification services was
not found.
InvalidKeyValue: The value of one of the keys is not valid
or the "Note about args" is violated.
not found.
InvalidKeyValue: The value of one of the keys is not valid or
the "Note about args" is violated.
Returns:
Reminder: The info about the reminder.
"""
from backend.features.reminder_handler import ReminderHandler
LOGGER.info(
f'Adding reminder with {title=}, {time=}, {notification_services=}, ' +
f'{text=}, {repeat_quantity=}, {repeat_interval=}, {weekdays=}, ' +
f'Adding reminder with {title=}, {time=}, {notification_services=}, '
f'{text=}, {repeat_quantity=}, {repeat_interval=}, {weekdays=}, '
f'{cron_schedule=}, {color=}, {enabled=}')
# Validate data
@@ -392,7 +415,8 @@ class Reminders:
):
raise InvalidKeyValue('weekdays', weekdays)
# Check if all notification services exist
# Check if all notification services exist. Creating an instance
# raises NotificationServiceNotFound if the ID is not valid.
for ns in notification_services:
NotificationService(self.user_id, ns)
@@ -419,8 +443,10 @@ class Reminders:
)
new_id = self.reminder_db.add(
title, text,
time, repeat_quantity_str,
title,
text,
time,
repeat_quantity_str,
repeat_interval,
weekdays_str,
cron_schedule,
@@ -430,9 +456,9 @@ class Reminders:
enabled
)
REMINDER_HANDLER.find_next_reminder(time)
ReminderHandler().find_next_reminder(time)
return self.fetchone(new_id)
return self.get_one(new_id)
def test_reminder(
self,
@@ -440,29 +466,31 @@ class Reminders:
notification_services: List[int],
text: str = ''
) -> SendResult:
"""Test send a reminder draft.
"""Test a reminder by sending it.
Args:
title (str): Title title of the entry.
title (str): The title of the reminder.
notification_service (int): The id of the notification service to
use to send the reminder.
notification_service (int): The IDs of the notification
services to use to send the reminder.
text (str, optional): The body of the reminder.
Defaults to ''.
Returns:
SendResult: Whether or not it was successful.
SendResult: The result of the test.
"""
LOGGER.info(
f'Testing reminder with {title=}, {notification_services=}, {text=}'
)
urls = [
NotificationService(self.user_id, ns_id).get().url
for ns_id in notification_services
]
return send_apprise_notification(
[
NotificationService(self.user_id, ns_id).get().url
for ns_id in notification_services
],
urls,
title,
text
)

4
backend/implementations/static_reminders.py
View File

@@ -177,7 +177,7 @@ class StaticReminders:
Defaults to TimelessSortingMethod.TITLE.
Returns:
List[StaticReminderData]: the info about all static reminders of
List[StaticReminderData]: The info about all static reminders of
the user that match the search term.
"""
return [
@@ -211,7 +211,7 @@ class StaticReminders:
"""Add a static reminder.
Args:
title (str): The title of the entry.
title (str): The title of the static reminder.
notification_services (List[int]): The IDs of the notification
services to use to send the static reminder.

8
frontend/api.py
View File

@@ -276,7 +276,7 @@ def api_reminders_list(inputs: Dict[str, Any]):
reminders = Reminders(api_key_map[g.hashed_api_key].user_data.user_id)
if request.method == 'GET':
result = reminders.fetchall(inputs['sort_by'])
result = reminders.get_all(inputs['sort_by'])
return return_api([r.todict() for r in result])
elif request.method == 'POST':
@@ -324,11 +324,11 @@ def api_get_reminder(inputs: Dict[str, Any], r_id: int):
)
if request.method == 'GET':
result = reminders.fetchone(r_id).get()
result = reminders.get_one(r_id).get()
return return_api(result.todict())
elif request.method == 'PUT':
result = reminders.fetchone(r_id).update(
result = reminders.get_one(r_id).update(
title=inputs['title'],
time=inputs['time'],
notification_services=inputs['notification_services'],
@@ -343,7 +343,7 @@ def api_get_reminder(inputs: Dict[str, Any], r_id: int):
return return_api(result.todict())
elif request.method == 'DELETE':
reminders.fetchone(r_id).delete()
reminders.get_one(r_id).delete()
return return_api({})