BE: direct DB access removed where possible

Signed-off-by: jokob-sk <jokob.sk@gmail.com>

front/plugins/csv_backup/script.py

@@ -4,7 +4,6 @@ import os
 import argparse
 import sys
 import csv
-import sqlite3
 from datetime import datetime
 
 # Register NetAlertX directories
@@ -13,9 +12,10 @@ sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
 
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value   # noqa: E402 [flake8 lint suppression]
-from const import logPath, fullDbPath  # noqa: E402 [flake8 lint suppression]
+from const import logPath  # noqa: E402 [flake8 lint suppression]
 import conf  # noqa: E402 [flake8 lint suppression]
 from pytz import timezone  # noqa: E402 [flake8 lint suppression]
+from database import get_temp_db_connection  # noqa: E402 [flake8 lint suppression]
 
 # Make sure the TIMEZONE for logging is correct
 conf.tz = timezone(get_setting_value('TIMEZONE'))
@@ -48,7 +48,7 @@ def main():
     mylog('verbose', ['[CSVBCKP] In script'])
 
     # Connect to the App database
-    conn = sqlite3.connect(fullDbPath)
+    conn = get_temp_db_connection()
     cursor = conn.cursor()
 
     # Execute your SQL query

front/plugins/sync/sync.py

@@ -4,7 +4,6 @@ import os
 import sys
 import requests
 import json
-import sqlite3
 import base64
 
 
@@ -15,13 +14,14 @@ sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
 from plugin_helper import Plugin_Objects  # noqa: E402 [flake8 lint suppression]
 from utils.plugin_utils import get_plugins_configs, decode_and_rename_files  # noqa: E402 [flake8 lint suppression]
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
-from const import fullDbPath, logPath  # noqa: E402 [flake8 lint suppression]
+from const import logPath  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value  # noqa: E402 [flake8 lint suppression]
 from utils.datetime_utils import timeNowDB  # noqa: E402 [flake8 lint suppression]
 from utils.crypto_utils import encrypt_data  # noqa: E402 [flake8 lint suppression]
 from messaging.in_app import write_notification  # noqa: E402 [flake8 lint suppression]
 import conf  # noqa: E402 [flake8 lint suppression]
 from pytz import timezone  # noqa: E402 [flake8 lint suppression]
+from database import get_temp_db_connection  # noqa: E402 [flake8 lint suppression]
 
 # Make sure the TIMEZONE for logging is correct
 conf.tz = timezone(get_setting_value('TIMEZONE'))
@@ -160,8 +160,8 @@ def main():
         mylog('verbose', [f'[{pluginName}] Mode 3: RECEIVE (HUB) - This is a HUB as received data found'])
 
         # Connect to the App database
-        conn    = sqlite3.connect(fullDbPath)
+        conn = get_temp_db_connection()
-        cursor  = conn.cursor()
+        cursor = conn.cursor()
 
         # Collect all unique devMac values from the JSON files
         unique_mac_addresses = set()

front/plugins/vendor_update/script.py

@@ -3,7 +3,6 @@
 import os
 import sys
 import subprocess
-import sqlite3
 
 # Register NetAlertX directories
 INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
@@ -12,10 +11,11 @@ sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
 from plugin_helper import Plugin_Objects, handleEmpty  # noqa: E402 [flake8 lint suppression]
 from logger import mylog, Logger  # noqa: E402 [flake8 lint suppression]
 from helper import get_setting_value   # noqa: E402 [flake8 lint suppression]
-from const import logPath, applicationPath, fullDbPath  # noqa: E402 [flake8 lint suppression]
+from const import logPath, applicationPath  # noqa: E402 [flake8 lint suppression]
 from scan.device_handling import query_MAC_vendor  # noqa: E402 [flake8 lint suppression]
 import conf  # noqa: E402 [flake8 lint suppression]
 from pytz import timezone  # noqa: E402 [flake8 lint suppression]
+from database import get_temp_db_connection  # noqa: E402 [flake8 lint suppression]
 
 # Make sure the TIMEZONE for logging is correct
 conf.tz = timezone(get_setting_value('TIMEZONE'))
@@ -40,7 +40,7 @@ def main():
     # Resolve missing vendors
     plugin_objects = Plugin_Objects(RESULT_FILE)
 
-    plugin_objects = update_vendors(fullDbPath, plugin_objects)
+    plugin_objects = update_vendors(plugin_objects)
 
     plugin_objects.write_result_file()
 
@@ -70,11 +70,11 @@ def update_vendor_database():
 
 # ------------------------------------------------------------------------------
 # resolve missing vendors
-def update_vendors(dbPath, plugin_objects):
+def update_vendors(plugin_objects):
 
     # Connect to the App SQLite database
-    conn = sqlite3.connect(dbPath)
+    conn = get_temp_db_connection()
-    sql  = conn.cursor()
+    cursor = conn.cursor()
 
     # Initialize variables
     ignored = 0
@@ -83,7 +83,7 @@ def update_vendors(dbPath, plugin_objects):
     mylog('verbose', ['    Searching devices vendor'])
 
     # Get devices without a vendor
-    sql.execute("""SELECT
+    cursor.execute("""SELECT
                             devMac,
                             devLastIP,
                             devName,
@@ -94,7 +94,7 @@ def update_vendors(dbPath, plugin_objects):
                                     OR devVendor   = ''
                                     OR devVendor   IS NULL
                         """)
-    devices = sql.fetchall()
+    devices = cursor.fetchall()
     conn.commit()
 
     # Close the database connection

scripts/db_empty/db_empty.py

@@ -1,4 +1,4 @@
-import sqlite3
+import sys
 import os
 
 # Connect to the database using environment variable
@@ -6,7 +6,14 @@ db_path = os.path.join(
     os.getenv('NETALERTX_DB', '/data/db'),
     'app.db'
 )
-conn = sqlite3.connect(db_path)
+
+# Register NetAlertX directories
+INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
+sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
+
+from database import get_temp_db_connection  # noqa: E402 [flake8 lint suppression]
+
+conn = get_temp_db_connection()
 cursor = conn.cursor()
 
 # Get the names of all tables (excluding SQLite internal tables)