add keyed parameters functionality

Author: siherrmann

database/db_job.py

@@ -95,11 +95,12 @@ def insert_job(self, job: Job) -> Job:
 
         with self.db.instance.cursor(row_factory=dict_row) as cur:
             cur.execute(
-                "SELECT * FROM insert_job(%s, %s, %s, %s, %s, %s);",
+                "SELECT * FROM insert_job(%s, %s, %s, %s, %s, %s, %s);",
                 (
                     json.dumps(job.options.to_dict()) if job.options else None,
                     job.task_name,
                     json.dumps(job.parameters),
+                    json.dumps(job.parameters_keyed),
                     job.status,
                     job.scheduled_at,
                     job.schedule_count,
@@ -144,21 +145,28 @@ def batch_insert_jobs(self, jobs: List[Job]) -> None:
             raise ValueError("Database connection is not established")
 
         try:
-            batch_data: List[Tuple[str, str, str, Optional[datetime]]] = []
+            batch_data: List[Tuple[str, str, str, str, Optional[datetime]]] = []
             for job in jobs:
                 options_json = (
                     json.dumps(job.options.to_dict()) if job.options else "{}"
                 )
                 parameters_json = json.dumps(job.parameters)
+                parameters_keyed_json = json.dumps(job.parameters_keyed)
                 batch_data.append(
-                    (options_json, job.task_name, parameters_json, job.scheduled_at)
+                    (
+                        options_json,
+                        job.task_name,
+                        parameters_json,
+                        parameters_keyed_json,
+                        job.scheduled_at,
+                    )
                 )
 
             with self.db.instance.cursor(row_factory=dict_row) as cur:
                 cur.executemany(
                     """
-                    INSERT INTO job (options, task_name, parameters, scheduled_at) 
-                    VALUES (%s, %s, %s, %s)
+                    INSERT INTO job (options, task_name, parameters, parameters_keyed, scheduled_at) 
+                    VALUES (%s, %s, %s, %s, %s)
                     """,
                     batch_data,
                 )

database/db_job_test.py

@@ -281,16 +281,23 @@ def test_update_stale_jobs(self):
 
     # DeleteJob
     def test_delete_job(self):
-        """Test deleting a job."""
+        """Test deleting a job from archive."""
         job_db_handler = JobDBHandler(self.db, with_table_drop=True)
 
         job = new_job("TestTask", None)
         inserted_job = job_db_handler.insert_job(job)
 
-        job_db_handler.delete_job(inserted_job.rid)
+        # Mark job as finished to move it to archive
+        inserted_job.status = "SUCCEEDED"
+        inserted_job.results = []
+        inserted_job.error = ""
+        archived_job = job_db_handler.update_job_final(inserted_job)
 
-        # Verify that the job no longer exists
-        deleted_job = job_db_handler.select_job(inserted_job.rid)
+        # Now delete from archive
+        job_db_handler.delete_job(archived_job.rid)
+
+        # Verify that the job no longer exists in archive
+        deleted_job = job_db_handler.select_job_from_archive(archived_job.rid)
         self.assertIsNone(deleted_job, "Expected deleted job to be None")
 
     # SelectJob

helper/logging_test.py

@@ -134,7 +134,7 @@ def test_error_logging_with_exception(self):
 
     def test_critical_logging(self):
         """Test critical level logging."""
-        logger = QueuerLogger(stream=self.stream)
+        logger = QueuerLogger(stream=self.stream, use_colors=False)
         logger.critical("Critical message")
 
         output = self.stream.getvalue()

model/batch_job.py

@@ -3,8 +3,8 @@
 Mirrors Go's model.BatchJob struct.
 """
 
-from typing import Any, List, Optional, Union, Callable
-from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Union, Callable
+from dataclasses import dataclass, field
 
 from .options import Options
 
@@ -16,5 +16,6 @@ class BatchJob:
     """
 
     task: Union[Callable[..., Any], str]
-    parameters: List[Any]
+    parameters: List[Any] = field(default_factory=list)
+    parameters_keyed: Dict[str, Any] = field(default_factory=dict)
     options: Optional[Options] = None

model/job.py

@@ -40,7 +40,8 @@ class Job:
 
     # Job definition
     task_name: str = ""
-    parameters: List[Any] = field(default_factory=lambda: [])
+    parameters: List[Any] = field(default_factory=list)
+    parameters_keyed: Dict[str, Any] = field(default_factory=dict)
     options: Optional[Options] = None
 
     # Job state
@@ -65,6 +66,7 @@ def to_dict(self) -> Dict[str, Any]:
             "worker_rid": str(self.worker_rid),
             "task_name": self.task_name,
             "parameters": self.parameters,
+            "parameters_keyed": self.parameters_keyed,
             "options": self.options.to_dict() if self.options else None,
             "status": self.status,
             "scheduled_at": (
@@ -110,6 +112,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "Job":
             job.worker_rid = UUID(data["worker_rid"])
         job.task_name = data.get("task_name", "")
         job.parameters = data.get("parameters", [])
+        job.parameters_keyed = data.get("parameters_keyed", {})
         if data.get("options"):
             job.options = Options.from_dict(data["options"])
         job.status = data.get("status", JobStatus.QUEUED)
@@ -173,6 +176,16 @@ def from_row(cls, row: Dict[str, Any]) -> "Job":
                 else parameters_value
             )
 
+        parameters_keyed_value = row.get(
+            "output_parameters_keyed", row.get("parameters_keyed")
+        )
+        if parameters_keyed_value:
+            job.parameters_keyed = (
+                json.loads(parameters_keyed_value)
+                if isinstance(parameters_keyed_value, str)
+                else parameters_keyed_value
+            )
+
         results_value = row.get("output_results", row.get("results"))
         if results_value is not None:
             # JSONB columns return Python objects directly, no JSON parsing needed
@@ -194,6 +207,7 @@ def new_job(
     task: Union[Callable[..., Any], str],
     options: Optional[Options] = None,
     *parameters: Any,
+    **parameters_keyed: Any,
 ) -> Job:
     """
     Create a new job from a task function or task name.
@@ -237,6 +251,7 @@ def new_job(
     job: Job = Job()
     job.task_name = task_name
     job.parameters = list(parameters)
+    job.parameters_keyed = parameters_keyed
     job.options = options
     job.status = status
     job.scheduled_at = scheduled_at

queuer_job.py

@@ -34,17 +34,23 @@ class QueuerJobMixin(QueuerGlobalMixin):
     def __init__(self):
         super().__init__()
 
-    def add_job(self, task: Union[Callable[..., Any], str], *parameters: Any) -> Job:
+    def add_job(
+        self,
+        task: Union[Callable[..., Any], str],
+        *parameters: Any,
+        **parameters_keyed: Any,
+    ) -> Job:
         """
         Add a job to the queue with the given task and parameters.
 
         :param task: Either a function or a string with the task name
-        :param parameters: Parameters to pass to the task
+        :param parameters: Positional parameters to pass to the task
+        :param parameters_keyed: Keyword parameters to pass to the task
         :returns: The created job
         :raises Exception: If something goes wrong
         """
         options: Optional[Options] = self._merge_options(None)
-        job: Job = self._add_job(task, options, *parameters)
+        job: Job = self._add_job(task, options, *parameters, **parameters_keyed)
 
         logger.info(f"Job added: {job.rid}")
 
@@ -55,14 +61,17 @@ def add_job_with_options(
         options: Optional[Options],
         task: Union[Callable[..., Any], str],
         *parameters: Any,
+        **parameters_keyed: Any,
     ) -> Job:
         """Add a new job with specific options."""
         # Merge default options with provided options
         options_merged: Optional[Options] = self._merge_options(options)
 
         try:
             # Create new job
-            new_job: Job = create_job(task, options_merged, *parameters)
+            new_job: Job = create_job(
+                task, options_merged, *parameters, **parameters_keyed
+            )
             job: Job = self.db_job.insert_job(new_job)
 
             logger.info(f"Job with options added: {job.rid}")
@@ -82,7 +91,9 @@ def add_jobs(self, batch_jobs: List[BatchJob]) -> None:
         for batch_job in batch_jobs:
             options: Optional[Options] = self._merge_options(batch_job.options)
             task_name: str = get_task_name_from_interface(batch_job.task)
-            job: Job = create_job(task_name, options, *batch_job.parameters)
+            job: Job = create_job(
+                task_name, options, *batch_job.parameters, **batch_job.parameters_keyed
+            )
             jobs.append(job)
 
         self.db_job.batch_insert_jobs(jobs)
@@ -232,7 +243,9 @@ def readd_job_from_archive(self, job_rid: UUID) -> Job:
             raise Exception(f"Job not found in archive: {job_rid}")
 
         # Readd the job to the queue
-        new_job = self.add_job_with_options(job.options, job.task_name, *job.parameters)
+        new_job = self.add_job_with_options(
+            job.options, job.task_name, *job.parameters, **job.parameters_keyed
+        )
         logger.info(f"Job readded: {new_job.rid}")
         return new_job
 
@@ -317,18 +330,20 @@ def _add_job(
         task: Union[Callable[..., Any], str],
         options: Optional["Options"],
         *parameters: Any,
+        **parameters_keyed: Any,
     ) -> Job:
         """
         Add a job to the queue with all necessary parameters.
 
         :param task: Either a function or a string with the task name
         :param options: Job-specific options (can be None)
-        :param parameters: Parameters to pass to the task
+        :param parameters: Positional parameters to pass to the task
+        :param parameters_keyed: Keyword parameters to pass to the task
         :returns: The created job
         :raises Exception: If something goes wrong
         """
         try:
-            new_job = create_job(task, options, *parameters)
+            new_job = create_job(task, options, *parameters, **parameters_keyed)
             job = self.db_job.insert_job(new_job)
 
             return job
@@ -405,7 +420,8 @@ async def _wait_for_job(
 
         try:
             parameters = getattr(job, "parameters", [])
-            runner = Runner(task, *parameters)
+            parameters_keyed = getattr(job, "parameters_keyed", {})
+            runner = Runner(task, *parameters, **parameters_keyed)
 
             logger.info(f"Created runner for job {job.rid}")