WIP: refactor, stage 2

.gitignore

@@ -1,6 +1,7 @@
 # Project-specific files
 external/
 log.txt
+*.db
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

app/application.py

@@ -4,7 +4,7 @@
     login_user,
     logout_user,
     login_required,
-    current_user
+    current_user,
 )
 from flask_sock import Sock
 from flask import Flask, render_template, jsonify, request, redirect, url_for
@@ -68,6 +68,7 @@
 
 # Admin routes
 
+
 class User(UserMixin):
     def __init__(self, username, password):
         self.username = ""
@@ -76,13 +77,16 @@ def __init__(self, username, password):
 
     def check_password(self, password):
         return True
+
     def is_authenticated(self):
         return True
-
+
+
 @login_manager.user_loader
 def load_user(user_id):
     return User("", "")
 
+
 @app.route("/init_app", methods=["GET", "POST"])
 def init_app_route():
     if request.method == "GET":
@@ -100,20 +104,21 @@ def init_app_route():
 
     return redirect(url_for("admin.login"))
 
+
 @app.route("/login", methods=["GET", "POST"])
 def login():
     if current_user.is_authenticated:
         return redirect(url_for("users"))
 
     if request.method == "GET":
         return render_template("auth/login.html")
-    
+
     username = request.form["username"]
     password = request.form["password"]
 
     user = User(username, password)
     if user is None or not user.check_password(password):
-        return render_template('auth/login.html', error='Invalid username or password')
+        return render_template("auth/login.html", error="Invalid username or password")
 
     login_user(user)
     return redirect(url_for("users"))

app/models/access_log.py

@@ -39,7 +39,7 @@ def get_full_log(cls, start_time=None, end_time=None, limit=100, offset=0):
             logs = query_event_logs(start_time='2023-01-01 00:00:00', end_time='2023-01-31 23:59:59',
             limit=50)
         """
-        connection = sqlite3.connect('database.db')
+        connection = sqlite3.connect("database.db")
         connection.row_factory = sqlite3.Row
         cursor = connection.cursor()
 
@@ -52,8 +52,10 @@ def get_full_log(cls, start_time=None, end_time=None, limit=100, offset=0):
 
         if start_time is not None and end_time is not None:
             query += "WHERE operation_time >= ? AND operation_time <= ?"
-            cursor.execute(query + " ORDER BY operation_time DESC LIMIT ? OFFSET ?",
-                           (start_time, end_time, limit, offset))
+            cursor.execute(
+                query + " ORDER BY operation_time DESC LIMIT ? OFFSET ?",
+                (start_time, end_time, limit, offset),
+            )
         else:
             query += "ORDER BY operation_time DESC LIMIT ? OFFSET ?"
             cursor.execute(query, (limit, offset))

app/models/device.py

@@ -18,75 +18,61 @@ def __init__(self, device_id, name, device_type, slack_channel_id):
         self.device_type = device_type
         self.slack_channel_id = slack_channel_id
 
-    @classmethod
-    def create_table(cls):
-        """Creates the devices table if it doesn't already exist."""
-        connection = sqlite3.connect('database.db')
-        cursor = connection.cursor()
-        cursor.execute('''
-            CREATE TABLE IF NOT EXISTS devices (
-                id TEXT PRIMARY KEY NOT NULL,
-                name TEXT NOT NULL,
-                type TEXT DEFAULT 'tool',
-                slack_channel_id TEXT DEFAULT NULL
-            )
-        ''')
-        connection.commit()
-        connection.close()
-
     @classmethod
     def get_all_devices(cls):
         """Fetches all devices from the database and returns them as dictionaries.
 
         Returns:
             List[dict]: A list of dictionaries representing the devices.
         """
-        connection = sqlite3.connect('database.db')
+        connection = sqlite3.connect("database.db")
         connection.row_factory = sqlite3.Row
         cursor = connection.cursor()
-        cursor.execute('SELECT * FROM devices')
+        cursor.execute("SELECT * FROM devices")
         devices_dict = cursor.fetchall()
-        
+
         # Convert the results to a list of dictionaries
         result_dicts = [dict(row) for row in devices_dict]
         connection.close()
         return result_dicts
-    
+
     @classmethod
     def get_devices_ids_and_names(cls):
-        #""Fetches all device ids and names
+        # ""Fetches all device ids and names
 
-        #Returns:
+        # Returns:
         #    List: A list of string of device names representing the devices.
-        #"
-        connection = sqlite3.connect('database.db')
+        # "
+        connection = sqlite3.connect("database.db")
         connection.row_factory = sqlite3.Row
         cursor = connection.cursor()
-        cursor.execute('SELECT id, name FROM devices')
+        cursor.execute("SELECT id, name FROM devices")
         devices = cursor.fetchall()
-        
+
         connection.close()
         device_dict = {}
         for device in devices:
-            device_id, device_name = device['id'], device['name']
+            device_id, device_name = device["id"], device["name"]
             device_dict[device_id] = device_name
 
         return device_dict
-    
+
     @classmethod
-
     def save(cls, device):
         """Saves a device to the database.
 
         Args:
             device (dict): A dictionary containing the device data.
         """
-        connection = sqlite3.connect('database.db')
+        connection = sqlite3.connect("database.db")
         cursor = connection.cursor()
-        cursor.execute('''
+        cursor.execute(
+            """
             INSERT INTO devices (id, name, type, slack_channel_id)
             VALUES (?, ?, ?, ?)
-        ''', (device['id'], device['name'], device['type'], device['slack_channel_id']))
+        """,
+            (device["id"], device["name"], device["type"], device["slack_channel_id"]),
+        )
         connection.commit()
         connection.close()
 
@@ -100,33 +86,33 @@ def get_by_id(cls, device_id):
         Returns:
             Device: The device with the specified ID, or None if not found.
         """
-        connection = sqlite3.connect('database.db')
+        connection = sqlite3.connect("database.db")
         cursor = connection.cursor()
-        cursor.execute('SELECT * FROM devices WHERE id = ?', (device_id,))
+        cursor.execute("SELECT * FROM devices WHERE id = ?", (device_id,))
         result = cursor.fetchone()
         if result:
             return Device(result[0], result[1], result[2], result[3])
         else:
             return None
+
     @classmethod
     def get_latest_key(cls):
         """
         Get last triggered key, to add new users by clicking on any reader
         """
-        connection = sqlite3.connect('database.db')
+        connection = sqlite3.connect("database.db")
         rows = (
             connection.cursor()
             .execute(
                 "SELECT user_key "
                 "FROM event_logs "
                 "WHERE user_key IS NOT NULL AND operation_type = 'deny_access' "
-                "ORDER BY operation_time DESC LIMIT 1")
+                "ORDER BY operation_time DESC LIMIT 1"
+            )
             .fetchone()
         )
         connection.close()
         if rows is None:
             return None
 
         return rows[0]
-
-print(Device.get_devices_ids_and_names())

app/models/user.py

@@ -1,10 +1,3 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-"""
-Created on Mon Nov 27 00:59:01 2023
-
-@author: artsin
-"""
 import sqlite3
 
 
@@ -14,35 +7,6 @@ def __init__(self, name, key, slack_id):
         self.key = key
         self.slack_id = slack_id
 
-    @classmethod
-    def create_table(cls):
-        connection = sqlite3.connect("database.db")
-        cursor = connection.cursor()
-        cursor.execute(
-            """
-            CREATE TABLE IF NOT EXISTS users (
-                name TEXT NOT NULL,
-                key TEXT NOT NULL UNIQUE,
-                slack_id TEXT DEFAULT NULL
-            )
-        """
-        )
-        connection.commit()
-        connection.close()
-
-    def save(self):
-        connection = sqlite3.connect("database.db")
-        cursor = connection.cursor()
-        cursor.execute(
-            """
-            INSERT INTO users (name, key, slack_id)
-            VALUES (?, ?, ?)
-        """,
-            (self.name, self.key, self.slack_id),
-        )
-        connection.commit()
-        connection.close()
-
     @classmethod
     def get_by_key(cls, key):
         connection = sqlite3.connect("database.db")
@@ -147,6 +111,3 @@ def remove_permission_for_device(self, device_id):
         connection.commit()
         connection.close()
 
-
-x = User.get_permissions()
-print(x)

app/schema.sql

@@ -0,0 +1,27 @@
+CREATE TABLE admins(
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        username TEXT NOT NULL,
+        password TEXT NOT NULL
+    );
+CREATE TABLE sqlite_sequence(name,seq);
+CREATE TABLE users(
+        name TEXT NOT NULL,
+        key  TEXT NOT NULL,
+        slack_id TEXT DEFAULT NULL
+    );
+CREATE TABLE permissions(
+        device_id TEXT NOT NULL,
+        user_key  TEXT NOT NULL
+    );
+CREATE TABLE event_logs(
+        device_id  TEXT NOT NULL,
+        user_key   TEXT,
+        operation_type TEXT NOT NULL,
+        operation_time       TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+    );
+CREATE TABLE IF NOT EXISTS "devices" (
+	"id"	TEXT NOT NULL,
+	"name"	TEXT NOT NULL,
+	"type"	TEXT DEFAULT "tool",
+	"slack_channel_id"	TEXT DEFAULT NULL
+);