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

Small database command optimisations

Browse Source
This commit is contained in:
CasVT
2023-05-22 21:56:43 +02:00
parent 054c7f798a
commit c60da4c0bc
7 changed files with 73 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/custom_exceptions.py
View File

@@ -2,6 +2,7 @@
from typing import Any, Dict
class UsernameTaken(Exception):
"""The username is already taken"""
api_response = {'error': 'UsernameTaken', 'result': {}, 'code': 400}

20
backend/db.py
View File

@@ -37,15 +37,15 @@ def get_db(output_type: Union[dict, tuple]=tuple):
Cursor: The Cursor instance to use
"""
try:
cursor = g.cursor
cursor = g.cursor
except AttributeError:
db = DBConnection(timeout=20.0)
cursor = g.cursor = db.cursor()
db = DBConnection(timeout=20.0)
cursor = g.cursor = db.cursor()
if output_type is dict:
cursor.row_factory = Row
cursor.row_factory = Row
else:
cursor.row_factory = None
cursor.row_factory = None
return g.cursor
@@ -73,10 +73,10 @@ def migrate_db(current_db_version: int) -> None:
# V1 -> V2
t = time()
utc_offset = datetime.fromtimestamp(t) - datetime.utcfromtimestamp(t)
reminders = cursor.execute("SELECT time, id FROM reminders;").fetchall()
cursor.execute("SELECT time, id FROM reminders;")
new_reminders = []
new_reminders_append = new_reminders.append
for reminder in reminders:
for reminder in cursor:
new_reminders_append([round((datetime.fromtimestamp(reminder[0]) - utc_offset).timestamp()), reminder[1]])
cursor.executemany("UPDATE reminders SET time = ? WHERE id = ?;", new_reminders)
current_db_version = 2
@@ -163,12 +163,14 @@ def setup_db() -> None:
""",
(__DATABASE_VERSION__,)
)
current_db_version = int(cursor.execute("SELECT value FROM config WHERE key = 'database_version' LIMIT 1;").fetchone()[0])
current_db_version = int(cursor.execute(
"SELECT value FROM config WHERE key = 'database_version' LIMIT 1;"
).fetchone()[0])
if current_db_version < __DATABASE_VERSION__:
migrate_db(current_db_version)
cursor.execute(
"UPDATE config SET value = ? WHERE key = 'database_version' LIMIT 1;",
"UPDATE config SET value = ? WHERE key = 'database_version';",
(__DATABASE_VERSION__,)
)

24
backend/notification_service.py
View File

@@ -8,12 +8,13 @@ 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",
"SELECT 1 FROM notification_services WHERE id = ? LIMIT 1;",
(self.id,)
).fetchone():
raise NotificationServiceNotFound
@@ -61,13 +62,15 @@ class NotificationService:
# Update database
get_db().execute("""
UPDATE notification_services
SET title=?, url=?
SET title = ?, url = ?
WHERE id = ?;
""", (
""",
(
data["title"],
data["url"],
self.id
))
)
)
return self.get()
@@ -103,17 +106,22 @@ class NotificationService:
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",
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

56
backend/reminders.py
View File

@@ -94,7 +94,7 @@ class ReminderHandler():
# Send of reminder
a = Apprise()
url = cursor.execute(
"SELECT url FROM notification_services WHERE id = ?",
"SELECT url FROM notification_services WHERE id = ? LIMIT 1;",
(reminder["notification_service"],)
).fetchone()["url"]
a.add(url)
@@ -102,7 +102,10 @@ class ReminderHandler():
if reminder['repeat_quantity'] is None:
# Delete the reminders from the database
cursor.execute("DELETE FROM reminders WHERE id = ?", (reminder['id'],))
cursor.execute(
"DELETE FROM reminders WHERE id = ?",
(reminder['id'],)
)
else:
# Set next time
new_time = _find_next_time(
@@ -138,7 +141,10 @@ class Reminder:
self.id = reminder_id
# Check if reminder exists
if not get_db().execute("SELECT 1 FROM reminders WHERE id = ? LIMIT 1", (self.id,)).fetchone():
if not get_db().execute(
"SELECT 1 FROM reminders WHERE id = ? LIMIT 1",
(self.id,)
).fetchone():
raise ReminderNotFound
def get(self) -> dict:
@@ -147,7 +153,7 @@ class Reminder:
Returns:
dict: The info about the reminder
"""
reminder: dict = get_db(dict).execute("""
reminder = get_db(dict).execute("""
SELECT
r.id,
r.title, r.text,
@@ -157,12 +163,11 @@ class Reminder:
r.repeat_quantity,
r.repeat_interval,
r.color
FROM
reminders r
INNER JOIN notification_services ns
ON
r.notification_service = ns.id
AND r.id = ?;
FROM reminders r
INNER JOIN notification_services ns
ON r.notification_service = ns.id
WHERE r.id = ?
LIMIT 1;
""",
(self.id,)
).fetchone()
@@ -229,7 +234,11 @@ class Reminder:
next_time = data["time"]
cursor.execute("""
UPDATE reminders
SET title=?, text=?, time=?, notification_service=?, repeat_quantity=?, repeat_interval=?, color=?
SET
title=?, text=?,
time=?, notification_service=?,
repeat_quantity=?, repeat_interval=?,
color=?
WHERE id = ?;
""", (
data["title"],
@@ -242,10 +251,17 @@ class Reminder:
self.id
))
else:
next_time = _find_next_time(data["time"], data["repeat_quantity"], data["repeat_interval"])
next_time = _find_next_time(
data["time"],
data["repeat_quantity"], data["repeat_interval"]
)
cursor.execute("""
UPDATE reminders
SET title=?, text=?, time=?, notification_service=?, repeat_quantity=?, repeat_interval=?, original_time=?, color=?
SET
title=?, text=?,
time=?, notification_service=?,
repeat_quantity=?, repeat_interval=?, original_time=?,
color=?
WHERE id = ?;
""", (
data["title"],
@@ -309,15 +325,13 @@ class Reminders:
r.repeat_quantity,
r.repeat_interval,
r.color
FROM
reminders r
INNER JOIN notification_services ns
ON
r.notification_service = ns.id
AND r.user_id = ?;
FROM reminders r
INNER JOIN notification_services ns
ON r.notification_service = ns.id
WHERE r.user_id = ?;
""",
(self.user_id,)
).fetchall()))
)))
# Sort result
reminders.sort(key=sort_function[0], reverse=sort_function[1])
@@ -418,7 +432,7 @@ def test_reminder(
"""
a = Apprise()
url = get_db(dict).execute(
"SELECT url FROM notification_services WHERE id = ?",
"SELECT url FROM notification_services WHERE id = ? LIMIT 1;",
(notification_service,)
).fetchone()
if not url:

1
backend/security.py
View File

@@ -5,6 +5,7 @@ from hashlib import pbkdf2_hmac
from secrets import token_bytes
from typing import Tuple
def get_hash(salt: bytes, data: str) -> bytes:
"""Hash a string using the supplied salt

11
backend/templates.py
View File

@@ -7,13 +7,17 @@ from backend.custom_exceptions import (NotificationServiceNotFound,
TemplateNotFound)
from backend.db import get_db
class Template:
"""Represents a template
"""
def __init__(self, template_id: int):
self.id = template_id
exists = get_db().execute("SELECT 1 FROM templates WHERE id = ? LIMIT 1;", (self.id,)).fetchone()
exists = get_db().execute(
"SELECT 1 FROM templates WHERE id = ? LIMIT 1;",
(self.id,)
).fetchone()
if not exists:
raise TemplateNotFound
@@ -30,7 +34,8 @@ class Template:
notification_service,
color
FROM templates
WHERE id = ?;
WHERE id = ?
LIMIT 1;
""",
(self.id,)
).fetchone()
@@ -113,7 +118,7 @@ class Templates:
ORDER BY title, id;
""",
(self.user_id,)
).fetchall()))
)))
return templates

2
backend/users.py
View File

@@ -17,7 +17,7 @@ class User:
def __init__(self, username: str, password: str):
# Fetch data of user to check if user exists and to check if password is correct
result = get_db(dict).execute(
"SELECT id, salt, hash FROM users WHERE username = ?",
"SELECT id, salt, hash FROM users WHERE username = ? LIMIT 1;",
(username,)
).fetchone()
if not result: