Replaced both DashboardLogger and PeerJobLogger with SqlAlchemy

src/modules/DashboardLogger.py

@@ -1,34 +1,40 @@
 """
 Dashboard Logger Class
 """
-import sqlite3, os, uuid
+import os, uuid
+import sqlalchemy as db
+from datetime import datetime
 
 class DashboardLogger:
     def __init__(self, CONFIGURATION_PATH):
-        self.loggerdb = sqlite3.connect(os.path.join(CONFIGURATION_PATH, 'db', 'wgdashboard_log.db'),
+        self.engine = db.create_engine(f'sqlite:///{os.path.join(CONFIGURATION_PATH, "db", "wgdashboard_log.db")}')
-                                        isolation_level=None,
+        self.loggerdb = self.engine.connect()
-                                        check_same_thread=False)
+        self.metadata = db.MetaData()
-        self.loggerdb.row_factory = sqlite3.Row
+        self.dashboardLoggerTable = db.Table('DashboardLog', self.metadata,
-        self.__createLogDatabase()
+                                             db.Column('LogID', db.VARCHAR, nullable=False, primary_key=True),
+                                             db.Column('LogDate', db.DATETIME,
+                                                       server_default=datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
+                                             db.Column('URL', db.VARCHAR),
+                                             db.Column('IP', db.VARCHAR),
+                                             
+                                             db.Column('Status', db.VARCHAR, nullable=False),
+                                             db.Column('Message', db.VARCHAR)
+                                             )
+        self.metadata.create_all(self.engine)
         self.log(Message="WGDashboard started")
-    def __createLogDatabase(self):
-        with self.loggerdb:
-            loggerdbCursor = self.loggerdb.cursor()
-            existingTable = loggerdbCursor.execute("SELECT name from sqlite_master where type='table'").fetchall()
-            existingTable = [t['name'] for t in existingTable]
-            if "DashboardLog" not in existingTable:
-                loggerdbCursor.execute(
-                    "CREATE TABLE DashboardLog (LogID VARCHAR NOT NULL, LogDate DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%S','now', 'localtime')), URL VARCHAR, IP VARCHAR, Status VARCHAR, Message VARCHAR, PRIMARY KEY (LogID))")
-            if self.loggerdb.in_transaction:
-                self.loggerdb.commit()
 
     def log(self, URL: str = "", IP: str = "", Status: str = "true", Message: str = "") -> bool:
         try:
-            loggerdbCursor = self.loggerdb.cursor()
+            with self.engine.begin() as conn:
-            loggerdbCursor.execute(
+                conn.execute(
-                "INSERT INTO DashboardLog (LogID, URL, IP, Status, Message) VALUES (?, ?, ?, ?, ?);", (str(uuid.uuid4()), URL, IP, Status, Message,))
+                    self.dashboardLoggerTable.insert().values(
-            loggerdbCursor.close()
+                        LogID=str(uuid.uuid4()),
-            self.loggerdb.commit()
+                        URL=URL,
+                        IP=IP,
+                        Status=Status,
+                        Message=Message
+                    )
+                )
             return True
         except Exception as e:
             print(f"[WGDashboard] Access Log Error: {str(e)}")

src/modules/PeerJobLogger.py

@@ -1,36 +1,40 @@
 """
 Peer Job Logger
 """
-import sqlite3, os, uuid
+import os, uuid
+import sqlalchemy as db
 from .Log import Log
+from datetime import datetime
 
 class PeerJobLogger:
     def __init__(self, CONFIGURATION_PATH, AllPeerJobs):
-        self.loggerdb = sqlite3.connect(os.path.join(CONFIGURATION_PATH, 'db', 'wgdashboard_log.db'),
+        self.engine = db.create_engine(f'sqlite:///{os.path.join(CONFIGURATION_PATH, "db", "wgdashboard_log.db")}')
-                                        check_same_thread=False)
+        self.loggerdb = self.engine.connect()
-        self.loggerdb.row_factory = sqlite3.Row
+        self.metadata = db.MetaData()
+        self.jobLogTable = db.Table('JobLog', self.metadata,
+                                    db.Column('LogID', db.VARCHAR, nullable=False, primary_key=True),
+                                    db.Column('JobID', db.VARCHAR, nullable=False),
+                                    db.Column('LogDate', db.DATETIME, 
+                                              server_default=datetime.now().strftime("%Y-%m-%d %H:%M:%S")),
+                                    db.Column('Status', db.VARCHAR, nullable=False),
+                                    db.Column('Message', db.VARCHAR)
+                                    )
         self.logs: list[Log] = []
-        self.__createLogDatabase()
+        self.metadata.create_all(self.engine)
         self.AllPeerJobs = AllPeerJobs
-    def __createLogDatabase(self):
-        with self.loggerdb:
-            loggerdbCursor = self.loggerdb.cursor()
-
-            existingTable = loggerdbCursor.execute("SELECT name from sqlite_master where type='table'").fetchall()
-            existingTable = [t['name'] for t in existingTable]
-
-            if "JobLog" not in existingTable:
-                loggerdbCursor.execute("CREATE TABLE JobLog (LogID VARCHAR NOT NULL, JobID NOT NULL, LogDate DATETIME DEFAULT (strftime('%Y-%m-%d %H:%M:%S','now', 'localtime')), Status VARCHAR NOT NULL, Message VARCHAR, PRIMARY KEY (LogID))")
-                if self.loggerdb.in_transaction:
-                    self.loggerdb.commit()
     def log(self, JobID: str, Status: bool = True, Message: str = "") -> bool:
         try:
-            with self.loggerdb:
+            with self.engine.begin() as conn:
-                loggerdbCursor = self.loggerdb.cursor()
+                conn.execute(
-                loggerdbCursor.execute(f"INSERT INTO JobLog (LogID, JobID, Status, Message) VALUES (?, ?, ?, ?)",
+                    self.jobLogTable.insert().values(
-                                       (str(uuid.uuid4()), JobID, Status, Message,))
+                        {
-                if self.loggerdb.in_transaction:
+                            "LogID": str(uuid.uuid4()), 
-                    self.loggerdb.commit()
+                            "JobID": JobID, 
+                            "Status": Status, 
+                            "Message": Message
+                        }
+                    )
+                )
         except Exception as e:
             print(f"[WGDashboard] Peer Job Log Error: {str(e)}")
             return False
@@ -40,14 +44,15 @@ class PeerJobLogger:
         logs: list[Log] = []
         try:
             allJobs = self.AllPeerJobs.getAllJobs(configName)
-            allJobsID = ", ".join([f"'{x.JobID}'" for x in allJobs])
+            allJobsID = [x.JobID for x in allJobs]
-            with self.loggerdb:
+            stmt = self.jobLogTable.select().where(self.jobLogTable.columns.JobID.in_(
-                loggerdbCursor = self.loggerdb.cursor()
+                allJobsID
-                table = loggerdbCursor.execute(f"SELECT * FROM JobLog WHERE JobID IN ({allJobsID}) ORDER BY LogDate DESC").fetchall()
+            ))
-                self.logs.clear()
+            table = self.loggerdb.execute(stmt).fetchall()            
-                for l in table:
+            for l in table:
-                    logs.append(
+                logs.append(
-                        Log(l["LogID"], l["JobID"], l["LogDate"], l["Status"], l["Message"]))
+                    Log(l.LogID, l.JobID, l.LogDate.strftime("%Y-%m-%d %H:%M:%S"), l.Status, l.Message))
         except Exception as e:
+            print(e)
             return logs
         return logs

src/requirements.txt

@@ -8,3 +8,4 @@ icmplib
 gunicorn
 requests
 tcconfig
+sqlalchemy