ClipSync/Server/Models/SQLite/sqlite.py

import sqlite3
import secrets
import time
from typing import Union,List
from Models.BaseModels.Clipboard import Clipboard
from Models.BaseModels.Notification import Notification
from Models.BaseModels.DB import BaseDb

class SQLite(BaseDb):

    def first_run(self) -> None:
        """
        create databases
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        cur.execute("CREATE TABLE USERS(\
                    username,\
                    password,\
                    token)")

        cur.execute("CREATE TABLE NOTIFICATION(\
                    id INTEGER PRIMARY KEY AUTOINCREMENT,\
                    date,\
                    machine,\
                    token,\
                    title,\
                    content)")

        cur.execute("CREATE TABLE CLIPBOARD(\
                    id INTEGER PRIMARY KEY AUTOINCREMENT,\
                    date,\
                    machine,\
                    token,\
                    content)")

        con.commit()
        con.close()

    ###############################################################################

    def authentificate_user(self,username, password) -> Union[bool, str]:
        """
        Authentifie un user, simple, basique, orelsan
        "username"
        "password"
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        db_user = cur.execute("SELECT token FROM USERS WHERE USERNAME=? AND \
                            PASSWORD=?", username, password).fetchone()
        con.close()
        if db_user is None:
            return None
        else:
            return db_user[0]

    def register(self,username, password) -> bool:
        """
        inscrit un user, simple, basique, orelsan
        "username"
        "password"
        """
        # check for username already existing
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        if cur.execute("SELECT username FROM USERS WHERE username=?", user)\
                .fetchone() is not None:
            con.close()
            return False, "Error, username already taken"
        # add user in db
        user = (username, password,
                secrets.token_urlsafe(64))
        try:
            cur.execute("INSERT INTO USERS VALUES (?,?,?)", user)
            con.commit()
            con.close()
            return True
        except:
            con.close()
            return False, "Error, most likely username already taken"

    ###############################################################################

    def add_notification(self,token, title, content, device_name) -> bool:
        """
        Ajoute une notification dans la DB
        """
        date = time.time()
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        if cur.execute("SELECT username FROM USERS WHERE token=?", (token,))\
                .fetchone() is None:
            return False, "Error, no user"
        try:
            notification = (date, device_name, token, title, content)
            cur.execute("INSERT INTO notification VALUES (null,?,?,?,?,?)",
                        notification)
            con.commit()
            con.close()

            return True
        except:
            return False, "error"

    def get_notification(self,token) -> Union[bool, List[Notification]]:
        """
        Trouve les notifications dans la DB
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        db_notifications = cur.execute("SELECT * \
                                    FROM NOTIFICATION \
                                    WHERE token=?",
                                    (token, )).fetchall()
        
        list_notifications = []
        for n in db_notifications:
            notif = Notification(*n)
            list_notifications.append(notif)

        return list_notifications

    def get_notification_by_id(self,token, id) -> Union[bool, Notification]:
        """
        Trouve la notification dans la DB avec son id
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        if id == -1:
            db_notifications = cur.execute("SELECT * \
                                        FROM NOTIFICATION \
                                        WHERE token=?\
                                        ORDER BY id DESC \
                                        LIMIT 1",
                                        (token,)).fetchone()
        else:
            db_notifications = cur.execute("SELECT * \
                                        FROM NOTIFICATION \
                                        WHERE token=? AND id=?",
                                        (token, id)).fetchone()
        
        list_notifications = []
        for n in db_notifications:
            notif = Notification(*n)
            list_notifications.append(notif)

        return list_notifications

    ###############################################################################

    def add_clipboard(self,token, content, device_name) -> bool:
        """
        Ajoute un clipboard dans la DB
        """
        date = time.time()
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        if cur.execute("SELECT username FROM USERS WHERE token=?", (token,))\
                .fetchone() is None:
            return False, "Error, no user"
        try:
            clipboard = (date, device_name, token, content)
            cur.execute("INSERT INTO CLIPBOARD VALUES (null,?,?,?,?)", clipboard)
            con.commit()
            con.close()
            return True
        except:
            return False, "error"

    def get_clipboard(self,token) -> Union[bool, List[Clipboard]]:
        """
        Trouves les clipboard dans la DB
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        db_clipboard = cur.execute("SELECT * \
                                FROM CLIPBOARD \
                                WHERE token=?",
                                (token, )).fetchall()
        
        list_clipboard = []
        for c in db_clipboard:
            clip = Clipboard(*c)
            list_clipboard.append(clip)

        return True, list_clipboard

    def get_clipboard_by_id(self,token, id) -> Union[bool, Clipboard]:
        """
        Trouve le clipbard dans le DB avec son id
        """
        con = sqlite3.connect("database.db")
        cur = con.cursor()
        if id == -1:
            db_clipboard = cur.execute("SELECT * \
                                    FROM CLIPBOARD \
                                    WHERE token=? \
                                    ORDER BY id DESC \
                                    LIMIT 1",
                                    (token,)).fetchone()
        else:
            db_clipboard = cur.execute("SELECT * \
                                    FROM CLIPBOARD \
                                    WHERE token=? AND id=?",
                                    (token, id)).fetchone()
            
        return True, Clipboard(*db_clipboard)