fix: database lock issues fixed using litequeue

nettacker/cli/db_worker.py

@@ -0,0 +1,47 @@
+import argparse
+import signal
+import time
+
+from nettacker.database.writer import get_writer
+
+
+def _handle_sig(signum, frame):
+    writer = get_writer()
+    writer.stop()
+    raise SystemExit(0)
+
+
+def run():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--once", action="store_true", help="Drain the queue once and exit")
+    parser.add_argument("--batch-size", type=int, default=None, help="Writer batch size")
+    parser.add_argument("--interval", type=float, default=None, help="Writer sleep interval")
+    parser.add_argument(
+        "--max-items", type=int, default=None, help="Max items to process in --once mode"
+    )
+    parser.add_argument("--summary", action="store_true", help="Print a summary after --once")
+    args = parser.parse_args()
+
+    signal.signal(signal.SIGINT, _handle_sig)
+    signal.signal(signal.SIGTERM, _handle_sig)
+
+    # apply runtime config
+    from nettacker.database.writer import get_writer_configured, get_stats
+
+    writer = get_writer_configured(batch_size=args.batch_size, interval=args.interval)
+    if args.once:
+        processed = writer.drain_once(max_iterations=args.max_items or 100000)
+        if args.summary:
+            stats = get_stats()
+            print(
+                f"processed={processed} total_processed={stats.get('processed')} queue_size={stats.get('queue_size')}"
+            )
+        return
+
+    # Main loop - will be terminated by signal handlers
+    while True:
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    run()

nettacker/database/db.py

@@ -9,6 +9,7 @@
 from nettacker.config import Config
 from nettacker.core.messages import messages
 from nettacker.database.models import HostsLog, Report, TempEvents
+from nettacker.database.writer import get_writer
 
 config = Config()
 log = logger.get_logger()
@@ -95,6 +96,22 @@ def submit_report_to_db(event):
         return True if submitted otherwise False
     """
     log.verbose_info(messages("inserting_report_db"))
+    writer = get_writer()
+    job = {
+        "action": "insert_report",
+        "payload": {
+            "date": event["date"],
+            "scan_id": event["scan_id"],
+            "report_path_filename": event["options"]["report_path_filename"],
+            "options": event["options"],
+        },
+    }
+    try:
+        if writer.enqueue(job):
+            return True
+    except Exception:
+        pass
+    # Fallback to direct write
     session = create_connection()
     session.add(
         Report(
@@ -140,6 +157,14 @@ def submit_logs_to_db(log):
         True if success otherwise False
     """
     if isinstance(log, dict):
+        writer = get_writer()
+        job = {"action": "insert_hostslog", "payload": log}
+        try:
+            if writer.enqueue(job):
+                return True
+        except Exception:
+            pass
+        # Fallback
         session = create_connection()
         session.add(
             HostsLog(
@@ -169,6 +194,14 @@ def submit_temp_logs_to_db(log):
         True if success otherwise False
     """
     if isinstance(log, dict):
+        writer = get_writer()
+        job = {"action": "insert_tempevent", "payload": log}
+        try:
+            if writer.enqueue(job):
+                return True
+        except Exception:
+            pass
+        # Fallback
         session = create_connection()
         session.add(
             TempEvents(