antinote/app/database.py

from uuid import uuid4

import mysql.connector

class Database:
    def __init__(self, app):
        self.db = mysql.connector.connect(
            host=app.config["DB_HOSTNAME"],
            user=app.config["DB_USERNAME"],
            password=app.config["DB_PASSWORD"],
            database=app.config["DB_DATABASE"]
        )

    def create_session(self, username, password):
        cursor = self.db.cursor()
        cursor.execute("SELECT user_id FROM utilisateurs WHERE username=%s AND password=%s", (username, password))

        if result := cursor.fetchall():
            user_id = result[0][0]
        else:
            return None

        uuid = str(uuid4())
        cursor.execute("INSERT INTO sessions (session_id, user_id) VALUES (%s, %s)", (uuid, user_id))
        self.db.commit()
        cursor.close()
        return uuid

    def destroy_session(self, uuid):
        cursor = self.db.cursor()
        cursor.execute("DELETE FROM sessions WHERE session_id=%s", (uuid,))
        self.db.commit()
        cursor.close()

    def check_connection(self, uuid):
        cursor = self.db.cursor()
        cursor.execute("DELETE FROM sessions WHERE expiry_date <= CURRENT_TIMESTAMP")
        cursor.execute("SELECT session_id FROM sessions WHERE session_id=%s", (uuid,))
        result = cursor.fetchall()
        cursor.close()
        return result

class User:
    def __init__(self, user_id, username, nom, prenom, creation_date):
        self.id = user_id
        self.username = username
        self.nom = nom
        self.prenom = prenom
        self.creation_date = creation_date

class Session:
    def __init__(self, db, uuid):
        self.db = db
        self.uuid = uuid
        self.user = self.get_user()

    def get_user(self):
        cursor = self.db.db.cursor()
        cursor.execute("SELECT utilisateurs.user_id, username, nom, prenom, creation_date FROM utilisateurs "
                       "JOIN sessions ON utilisateurs.user_id=sessions.user_id WHERE session_id=%s", (self.uuid,))
        result = cursor.fetchall()
        cursor.close()
        return User(*result[0])

    def fetch_grades(self):
        cursor = self.db.db.cursor()
        cursor.execute("SELECT matieres.libelle, devoirs.libelle, date, coefficient, valeur, maximum FROM notes "
                       "JOIN devoirs ON notes.id_devoir=devoirs.id_devoir "
                       "JOIN matieres ON devoirs.id_matiere=matieres.id_matiere WHERE id_eleve=%s", (self.user.id,))

        result = cursor.fetchall()
        cursor.close()
        dico = {}
        for note in result:
            if note[0] in dico:
                dico[note[0]].append(note)
            else:
                dico[note[0]] = [note]
        return dico