refactor: improve ingester typing

This will be handy for checking which objects to update in the denormalized tables

Author: MarceloRobert

backend/.gitignore

@@ -180,3 +180,6 @@ runtime/secrets/postgres_password_secret
 # This is where the sqlite cache will be stored by default if not running on Docker.
 volume_data/*.sqlite3
 volume_data/*.yaml
+
+# Temporary submissions directory (for development/testing purposes)
+kernelCI_app/management/commands/tmp_submissions/*

backend/kernelCI_app/management/commands/helpers/kcidbng_ingester.py

@@ -17,6 +17,7 @@
 from kernelCI_app.models import Issues, Checkouts, Builds, Tests, Incidents
 
 from kernelCI_app.management.commands.helpers.process_submissions import (
+    TableNames,
     build_instances_from_submission,
 )
 
@@ -293,6 +294,38 @@ def prepare_file_data(filename, trees_name, spool_dir):
         }
 
 
+# TODO: MOVE THESE TYPES AND CONSTS
+type TableModels = Issues | Checkouts | Builds | Tests | Incidents
+
+MODEL_MAP: dict[TableNames, TableModels] = {
+    "issues": Issues,
+    "checkouts": Checkouts,
+    "builds": Builds,
+    "tests": Tests,
+    "incidents": Incidents,
+}
+
+
+def consume_buffer(buffer: list[TableModels], item_type: TableNames) -> None:
+    """
+    Consume a buffer of items and insert them into the database.
+    This function is called by the db_worker thread.
+    """
+    if not buffer:
+        return
+
+    model = MODEL_MAP[item_type]
+
+    t0 = time.time()
+    model.objects.bulk_create(
+        buffer,
+        batch_size=INGEST_BATCH_SIZE,
+        ignore_conflicts=True,
+    )
+    _out("bulk_create %s: n=%d in %.3fs" % (item_type, len(buffer), time.time() - t0))
+
+
+# TODO: lower the complexity of this function
 def db_worker(stop_event: threading.Event):  # noqa: C901
     """
     Worker thread that processes the database queue.
@@ -303,11 +336,11 @@ def db_worker(stop_event: threading.Event):  # noqa: C901
     """
 
     # Local buffers for batching
-    issues_buf = []
-    checkouts_buf = []
-    builds_buf = []
-    tests_buf = []
-    incidents_buf = []
+    issues_buf: list[Issues] = []
+    checkouts_buf: list[Checkouts] = []
+    builds_buf: list[Builds] = []
+    tests_buf: list[Tests] = []
+    incidents_buf: list[Incidents] = []
 
     last_flush_ts = time.time()
 
@@ -331,55 +364,11 @@ def flush_buffers():
         try:
             # Single transaction for all tables in the flush
             with transaction.atomic():
-                if issues_buf:
-                    t0 = time.time()
-                    Issues.objects.bulk_create(
-                        issues_buf, batch_size=INGEST_BATCH_SIZE, ignore_conflicts=True
-                    )
-                    _out(
-                        "bulk_create issues: n=%d in %.3fs"
-                        % (len(issues_buf), time.time() - t0)
-                    )
-                if checkouts_buf:
-                    t0 = time.time()
-                    Checkouts.objects.bulk_create(
-                        checkouts_buf,
-                        batch_size=INGEST_BATCH_SIZE,
-                        ignore_conflicts=True,
-                    )
-                    _out(
-                        "bulk_create checkouts: n=%d in %.3fs"
-                        % (len(checkouts_buf), time.time() - t0)
-                    )
-                if builds_buf:
-                    t0 = time.time()
-                    Builds.objects.bulk_create(
-                        builds_buf, batch_size=INGEST_BATCH_SIZE, ignore_conflicts=True
-                    )
-                    _out(
-                        "bulk_create builds: n=%d in %.3fs"
-                        % (len(builds_buf), time.time() - t0)
-                    )
-                if tests_buf:
-                    t0 = time.time()
-                    Tests.objects.bulk_create(
-                        tests_buf, batch_size=INGEST_BATCH_SIZE, ignore_conflicts=True
-                    )
-                    _out(
-                        "bulk_create tests: n=%d in %.3fs"
-                        % (len(tests_buf), time.time() - t0)
-                    )
-                if incidents_buf:
-                    t0 = time.time()
-                    Incidents.objects.bulk_create(
-                        incidents_buf,
-                        batch_size=INGEST_BATCH_SIZE,
-                        ignore_conflicts=True,
-                    )
-                    _out(
-                        "bulk_create incidents: n=%d in %.3fs"
-                        % (len(incidents_buf), time.time() - t0)
-                    )
+                consume_buffer(issues_buf, "issues")
+                consume_buffer(checkouts_buf, "checkouts")
+                consume_buffer(builds_buf, "builds")
+                consume_buffer(tests_buf, "tests")
+                consume_buffer(incidents_buf, "incidents")
         except Exception as e:
             logger.error("Error during bulk_create flush: %s", e)
         finally:

backend/kernelCI_app/management/commands/helpers/process_submissions.py

@@ -1,14 +1,26 @@
 import logging
 from django.utils import timezone
-from typing import Any, Literal
+from typing import Any, Literal, TypedDict
 
 from django.db import IntegrityError
 from pydantic import ValidationError
 
 from kernelCI_app.models import Builds, Checkouts, Incidents, Issues, Tests
 
 
-TableNames = Literal["issues", "checkouts", "builds", "tests", "incidents"]
+# TODO: move to a types file/folder
+type TableNames = Literal["issues", "checkouts", "builds", "tests", "incidents"]
+
+
+class ProcessedSubmission(TypedDict):
+    """Stores the list of items in a single submission.
+    Lists can't be None but can be empty."""
+
+    issues: list[Issues]
+    checkouts: list[Checkouts]
+    builds: list[Builds]
+    tests: list[Tests]
+    incidents: list[Incidents]
 
 
 logger = logging.getLogger(__name__)
@@ -128,12 +140,12 @@ def make_incident_instance(incident) -> Incidents:
     return obj
 
 
-def build_instances_from_submission(data: dict[str, Any]) -> dict[TableNames, list]:
+def build_instances_from_submission(data: dict[str, Any]) -> ProcessedSubmission:
     """
     Convert raw submission dicts into unsaved Django model instances, grouped by type.
     Per-item errors are logged and the item is skipped, matching the previous behavior.
     """
-    out: dict[TableNames, list] = {
+    out: ProcessedSubmission = {
         "issues": [],
         "checkouts": [],
         "builds": [],