mirror of
https://github.com/Casvt/MIND.git
synced 2026-02-19 11:54:46 -05:00
155 lines
3.6 KiB
Python
155 lines
3.6 KiB
Python
#-*- coding: utf-8 -*-
|
|
|
|
from typing import List
|
|
|
|
from apprise import Apprise
|
|
|
|
from backend.custom_exceptions import (InvalidURL, NotificationServiceInUse,
|
|
NotificationServiceNotFound)
|
|
from backend.db import get_db
|
|
|
|
class NotificationService:
|
|
def __init__(self, notification_service_id: int) -> None:
|
|
self.id = notification_service_id
|
|
|
|
if not get_db().execute(
|
|
"SELECT 1 FROM notification_services WHERE id = ? LIMIT 1",
|
|
(self.id,)
|
|
).fetchone():
|
|
raise NotificationServiceNotFound
|
|
|
|
def get(self) -> dict:
|
|
"""Get the info about the notification service
|
|
|
|
Returns:
|
|
dict: The info about the notification service
|
|
"""
|
|
result = dict(get_db(dict).execute(
|
|
"SELECT id, title, url FROM notification_services WHERE id = ? LIMIT 1",
|
|
(self.id,)
|
|
).fetchone())
|
|
|
|
return result
|
|
|
|
def update(
|
|
self,
|
|
title: str = None,
|
|
url: str = None
|
|
) -> dict:
|
|
"""Edit the notification service
|
|
|
|
Args:
|
|
title (str, optional): The new title of the service. Defaults to None.
|
|
url (str, optional): The new url of the service. Defaults to None.
|
|
|
|
Returns:
|
|
dict: The new info about the service
|
|
"""
|
|
if not Apprise().add(url):
|
|
raise InvalidURL
|
|
|
|
# Get current data and update it with new values
|
|
data = self.get()
|
|
new_values = {
|
|
'title': title,
|
|
'url': url
|
|
}
|
|
for k, v in new_values.items():
|
|
if v is not None:
|
|
data[k] = v
|
|
|
|
# Update database
|
|
get_db().execute("""
|
|
UPDATE notification_services
|
|
SET title=?, url=?
|
|
WHERE id = ?;
|
|
""", (
|
|
data["title"],
|
|
data["url"],
|
|
self.id
|
|
))
|
|
|
|
return self.get()
|
|
|
|
def delete(self) -> None:
|
|
"""Delete the service
|
|
|
|
Raises:
|
|
NotificationServiceInUse: The service is still used by a reminder
|
|
"""
|
|
# Check if no reminders exist with this service
|
|
cursor = get_db()
|
|
cursor.execute(
|
|
"SELECT 1 FROM reminders WHERE notification_service = ? LIMIT 1;",
|
|
(self.id,)
|
|
)
|
|
if cursor.fetchone():
|
|
raise NotificationServiceInUse('reminder')
|
|
|
|
# Check if no templates exist with this service
|
|
cursor.execute(
|
|
"SELECT 1 FROM templates WHERE notification_service = ? LIMIT 1;",
|
|
(self.id,)
|
|
)
|
|
if cursor.fetchone():
|
|
raise NotificationServiceInUse('template')
|
|
|
|
cursor.execute(
|
|
"DELETE FROM notification_services WHERE id = ?",
|
|
(self.id,)
|
|
)
|
|
return
|
|
|
|
class NotificationServices:
|
|
def __init__(self, user_id: int) -> None:
|
|
self.user_id = user_id
|
|
|
|
def fetchall(self) -> List[dict]:
|
|
"""Get a list of all notification services
|
|
|
|
Returns:
|
|
List[dict]: The list of all notification services
|
|
"""
|
|
result = list(map(dict, get_db(dict).execute(
|
|
"SELECT id, title, url FROM notification_services WHERE user_id = ? ORDER BY title, id",
|
|
(self.user_id,)
|
|
).fetchall()))
|
|
|
|
return result
|
|
|
|
def fetchone(self, notification_service_id: int) -> NotificationService:
|
|
"""Get one notification service based on it's id
|
|
|
|
Args:
|
|
notification_service_id (int): The id of the desired service
|
|
|
|
Returns:
|
|
NotificationService: Instance of NotificationService
|
|
"""
|
|
return NotificationService(notification_service_id)
|
|
|
|
def add(self, title: str, url: str) -> NotificationService:
|
|
"""Add a notification service
|
|
|
|
Args:
|
|
title (str): The title of the service
|
|
url (str): The apprise url of the service
|
|
|
|
Raises:
|
|
InvalidURL: The apprise url is invalid
|
|
|
|
Returns:
|
|
dict: The info about the new service
|
|
"""
|
|
if not Apprise().add(url):
|
|
raise InvalidURL
|
|
|
|
new_id = get_db().execute("""
|
|
INSERT INTO notification_services(user_id, title, url)
|
|
VALUES (?,?,?)
|
|
""",
|
|
(self.user_id, title, url)
|
|
).lastrowid
|
|
|
|
return self.fetchone(new_id)
|
|
|