upsert tests and mark complete in bulk

server/db_py.py

@@ -104,9 +104,12 @@ def del_config(key):
         _conn.execute("DELETE FROM config WHERE key = ?", (key,))
 
 
-def upsert_test(test):
+def upsert_tests(tests):
+    if not tests:
+        return
+
     with _tx():
-        _conn.execute(
+        _conn.executemany(
             """
             INSERT INTO tests
                 (id, test_id, parent_dir, filename, interference, device, rotation,
@@ -130,14 +133,29 @@ def upsert_test(test):
                 direction    = excluded.direction,
                 throttled    = excluded.throttled
             """,
-            test,
+            tests,
         )
 
 
-def mark_completed(test_id, device, completed_at, duration_seconds, tput_results=None):
+def mark_tests_completed(records):
-    json_value = json.dumps(tput_results) if tput_results else None
+    if not records:
+        return
+
+    payload = []
+    for record in records:
+        payload.append(
+            (
+                record.get("completed_at"),
+                record.get("duration_seconds"),
+                json.dumps(record.get("tput_results")) if record.get("tput_results") else None,
+                record.get("test_id"),
+                record.get("device"),
+                record.get("device"),
+            )
+        )
+
     with _tx():
-        _conn.execute(
+        _conn.executemany(
             """
             UPDATE tests
             SET completed = 1,
@@ -147,7 +165,7 @@ def mark_completed(test_id, device, completed_at, duration_seconds, tput_results
             WHERE test_id = ?
               AND (? IS NULL OR device = ?)
             """,
-            (completed_at, duration_seconds, json_value, test_id, device, device),
+            payload,
         )
 
 

server/scanner.py

@@ -5,14 +5,14 @@ import tempfile
 import smbclient
 
 from db_py import (
+    mark_tests_completed,
+    upsert_tests,
     clear_tests,
     extract_measurement_metrics,
     get_config,
-    mark_completed,
     reset_all_results_state,
     update_all_p2p_coe_pairs_sql,
     update_all_p3p_pairs_sql,
-    upsert_test,
 )
 from parser import (
     parse_result_filename,
@@ -335,6 +335,7 @@ def full_scan(target_dir, results_dir, results_dir_ref):
 def scan_targets(target_dir):
     target_dir = _normalize_input_path(target_dir)
     exclusions = _parse_scan_exclusions(get_config("scan_exclusions"))
+    batch_tests = []
 
     try:
         parent_entries = [
@@ -378,7 +379,7 @@ def scan_targets(target_dir):
                 continue
 
             #print( f"[scanner] found: {parent_name}/{filename}  ->  test_id={parsed['test_id']}")
-            upsert_test(
+            batch_tests.append(
                 {
                     "id": parsed["id"],
                     "test_id": parsed["test_id"],
@@ -398,6 +399,8 @@ def scan_targets(target_dir):
                 }
             )
 
+    upsert_tests(batch_tests)
+
     return True
 
 
@@ -415,8 +418,13 @@ def scan_results(results_dir):
         return
 
     print(f"[scanner] results: {len(entries)} result dir(s) found")
+    completed_batch = []
     for dir_name in entries:
-        process_result_dir(results_dir, dir_name)
+        completion = process_result_dir(results_dir, dir_name)
+        if completion:
+            completed_batch.append(completion)
+
+    mark_tests_completed(completed_batch)
 
 
 def process_result_dir(results_dir, result_dir_name):
@@ -426,7 +434,7 @@ def process_result_dir(results_dir, result_dir_name):
     test_id = parsed["test_id"]
     device = parsed["device"]
     if not test_id or not device:
-        return
+        return None
 
     result_dir_path = _join_path(results_dir, result_dir_name)
 
@@ -434,13 +442,18 @@ def process_result_dir(results_dir, result_dir_name):
         files = [entry.name for entry in _iter_dir_entries(result_dir_path) if entry.is_file()]
     except OSError as exc:
         print(f"[scanner] Cannot read result dir {result_dir_name}: {exc}")
-        return
+        return None
 
     measurement_db = next((name for name in files if name.lower() == "measurement.db"), None)
     if not measurement_db:
         print(f"[scanner] completed (no measurement.db yet): {test_id}")
-        mark_completed(test_id, device, None, None)
+        return {
-        return
+            "test_id": test_id,
+            "device": device,
+            "completed_at": None,
+            "duration_seconds": None,
+            "tput_results": None,
+        }
 
     db_path = _join_path(result_dir_path, measurement_db)
     completed_at = parse_timestamp(result_dir_name)
@@ -450,13 +463,13 @@ def process_result_dir(results_dir, result_dir_name):
         f"[scanner] completed: {test_id}  device={device}  "
         f"duration={data['duration_seconds']}s  stations={len(data['tputResults'])}  at={completed_at}"
     )'''
-    mark_completed(
+    return {
-        test_id,
+        "test_id": test_id,
-        device,
+        "device": device,
-        completed_at,
+        "completed_at": completed_at,
-        data["duration_seconds"],
+        "duration_seconds": data["duration_seconds"],
-        data["tputResults"],
+        "tput_results": data["tputResults"],
-    )
+    }
 
 
 def parse_deleted_result_dir_name(dir_name):