New jobs API for accessing the ML task queue and posting back results (#1046)

Defines an initial set of API endpoints for an external processing service to read & pull ML tasks from the job queue and post back results (detections & predictions).

- Add tasks() endpoint (stubbed) for NATS JetStream integration
- Add result() endpoint (stubbed) for pipeline result processing
- Add ids_only and incomplete_only query parameters to job list endpoint
- Add JobFilterSet with pipeline__slug filtering and search fields

- Add utility OpenApiParameter definitions for new query params

The tasks() and result() endpoints are documented and stubbed out,
awaiting NATS and Celery task dependencies from PR #987.

* test: Add comprehensive unit tests for new JobViewSet features

- test_list_jobs_with_ids_only: Verifies ids_only parameter returns job IDs
- test_list_jobs_with_incomplete_only: Tests incomplete_only filtering by results stage
- test_filter_by_pipeline_slug: Tests JobFilterSet pipeline__slug filtering
- test_search_jobs: Tests search functionality by name and pipeline name
- test_tasks_endpoint_stub: Tests stubbed tasks() endpoint
- test_tasks_endpoint_with_batch: Tests batch parameter handling
- test_tasks_endpoint_without_pipeline: Tests validation for missing pipeline
- test_result_endpoint_stub: Tests stubbed result() endpoint
- test_result_endpoint_validation: Tests request validation for result()

All tests pass successfully, verifying the new API features work correctly.

* refactor: Use helper methods in JobViewSet tests for cleaner code

* Update ami/jobs/views.py

CR feedback

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update ami/jobs/views.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update ami/jobs/views.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Use self.get_object()

* Lean out implementation

* Use pydantic types for tasks/results

* Update tests

* coderabbit feedback

* Better batch validation

* better tests, fix formatting settings

* Fix typo

* Move filtering to filter_queryset

* Undo settings change

* Make jobs/tasks?batch required

* CR nits

* Refactor

* Update tests

* chore: move openAPI param definitions to job/schemas

* docs: add inline todos

* chore: move project openAPI param to main/schemas

* test: fix periodically failing test due to pagination. update asserts.

* fix: add missing new schema files

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Michael Bunsen <notbot@gmail.com>

Author: 3 people

ami/jobs/schemas.py

@@ -0,0 +1,22 @@
+from drf_spectacular.utils import OpenApiParameter
+
+ids_only_param = OpenApiParameter(
+    name="ids_only",
+    description="Return only job IDs instead of full objects",
+    required=False,
+    type=bool,
+)
+
+incomplete_only_param = OpenApiParameter(
+    name="incomplete_only",
+    description="Filter to only incomplete jobs (excludes jobs with final state in 'results' stage)",
+    required=False,
+    type=bool,
+)
+
+batch_param = OpenApiParameter(
+    name="batch",
+    description="Number of tasks to retrieve",
+    required=False,
+    type=int,
+)

ami/jobs/serializers.py

@@ -151,3 +151,12 @@ class Meta(JobListSerializer.Meta):
         fields = JobListSerializer.Meta.fields + [
             "result",
         ]
+
+
+class MinimalJobSerializer(DefaultSerializer):
+    """Minimal serializer returning only essential job fields."""
+
+    class Meta:
+        model = Job
+        # Add other fields when needed, e.g:  "name", "status", "created_at"
+        fields = ["id"]

ami/jobs/tests.py

@@ -1,5 +1,6 @@
 # from rich import print
 import logging
+from typing import Any
 
 from django.test import TestCase
 from guardian.shortcuts import assign_perm
@@ -88,6 +89,7 @@ def setUp(self):
             is_superuser=True,
         )
         self.factory = APIRequestFactory()
+        self.pipeline = None  # type: Pipeline | None
 
     def test_get_job(self):
         self.client.force_authenticate(user=self.user)
@@ -134,6 +136,30 @@ def _create_job(self, name: str, start_now: bool = True):
         self.assertEqual(resp.status_code, 201)
         return resp.json()
 
+    def _create_pipeline(self, name: str = "Test Pipeline", slug: str = "test-pipeline") -> Pipeline:
+        """Helper to create a pipeline and add it to the project."""
+        if self.pipeline and self.pipeline.slug == slug and self.pipeline.name == name:
+            return self.pipeline
+
+        pipeline = Pipeline.objects.create(
+            name=name,
+            slug=slug,
+            description=f"{name} description",
+        )
+        pipeline.projects.add(self.project)
+        self.pipeline = pipeline
+        return pipeline
+
+    def _create_ml_job(self, name: str, pipeline: Pipeline) -> Job:
+        """Helper to create an ML job with a pipeline."""
+        return Job.objects.create(
+            job_type_key=MLJob.key,
+            project=self.project,
+            name=name,
+            pipeline=pipeline,
+            source_image_collection=self.source_image_collection,
+        )
+
     def test_create_job(self):
         job_name = "Test job - Start but don't run"
         data = self._create_job(job_name, start_now=False)
@@ -201,3 +227,190 @@ def test_cancel_job(self):
         # This cannot be tested until we have a way to cancel jobs
         # and a way to run async tasks in tests.
         pass
+
+    def test_list_jobs_with_ids_only(self):
+        """Test the ids_only parameter returns only job IDs."""
+        # Create additional jobs via API
+        self._create_job("Test job 2", start_now=False)
+        self._create_job("Test job 3", start_now=False)
+
+        self.client.force_authenticate(user=self.user)
+        jobs_list_url = reverse_with_params("api:job-list", params={"project_id": self.project.pk, "ids_only": True})
+        resp = self.client.get(jobs_list_url)
+
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertIn("count", data)
+        self.assertEqual(data["count"], 3)  # Original job + 2 new ones
+        self.assertEqual(len(data["results"]), 3)
+        # Verify these are actually IDs
+        self.assertTrue(all(isinstance(r["id"], int) for r in data["results"]))
+        # Verify we don't get the full results structure
+        self.assertNotIn("details", data["results"][0])
+
+    def test_list_jobs_with_incomplete_only(self):
+        """Test the incomplete_only parameter filters jobs correctly."""
+        # Create jobs via API
+        completed_data = self._create_job("Completed job", start_now=False)
+        incomplete_data = self._create_job("Incomplete job", start_now=False)
+
+        # Mark completed job as complete by setting results stage to SUCCESS
+        completed_job = Job.objects.get(pk=completed_data["id"])
+        completed_job.progress.add_stage("results")
+        completed_job.progress.update_stage("results", progress=1.0, status=JobState.SUCCESS)
+        completed_job.save()
+
+        # Mark incomplete job as incomplete
+        incomplete_job = Job.objects.get(pk=incomplete_data["id"])
+        incomplete_job.progress.add_stage("results")
+        incomplete_job.progress.update_stage("results", progress=0.5, status=JobState.STARTED)
+        incomplete_job.save()
+
+        self.client.force_authenticate(user=self.user)
+        jobs_list_url = reverse_with_params(
+            "api:job-list", params={"project_id": self.project.pk, "incomplete_only": True}
+        )
+        resp = self.client.get(jobs_list_url)
+
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        # Should only return the incomplete job and the original test job (which has no results stage)
+        returned_ids = [job["id"] for job in data["results"]]
+        self.assertIn(incomplete_job.pk, returned_ids)
+        self.assertIn(self.job.pk, returned_ids)  # Original job has no results stage
+        self.assertNotIn(completed_job.pk, returned_ids)
+
+    def test_filter_by_pipeline_slug(self):
+        """Test filtering jobs by pipeline__slug."""
+        pipeline = self._create_pipeline("Test Pipeline", "test-pipeline")
+        job_with_pipeline = self._create_ml_job("Job with pipeline", pipeline)
+
+        self.client.force_authenticate(user=self.user)
+        jobs_list_url = reverse_with_params(
+            "api:job-list", params={"project_id": self.project.pk, "pipeline__slug": "test-pipeline"}
+        )
+        resp = self.client.get(jobs_list_url)
+
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["id"], job_with_pipeline.pk)
+
+    def test_search_jobs(self):
+        """Test searching jobs by name and pipeline name."""
+        pipeline = self._create_pipeline("SearchablePipeline", "searchable-pipeline")
+
+        self._create_ml_job("Find me job", pipeline)
+        self._create_job("Other job", start_now=False)
+
+        self.client.force_authenticate(user=self.user)
+
+        # Search by job name
+        jobs_list_url = reverse_with_params("api:job-list", params={"project_id": self.project.pk, "search": "Find"})
+        resp = self.client.get(jobs_list_url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data["count"], 1)
+        self.assertIn("Find me", data["results"][0]["name"])
+
+        # Search by pipeline name
+        jobs_list_url = reverse_with_params(
+            "api:job-list", params={"project_id": self.project.pk, "search": "Searchable"}
+        )
+        resp = self.client.get(jobs_list_url)
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data["count"], 1)
+        self.assertEqual(data["results"][0]["pipeline"]["name"], "SearchablePipeline")
+
+    def _task_batch_helper(self, value: Any, expected_status: int):
+        pipeline = self._create_pipeline()
+        job = self._create_ml_job("Job for batch test", pipeline)
+
+        self.client.force_authenticate(user=self.user)
+        tasks_url = reverse_with_params(
+            "api:job-tasks", args=[job.pk], params={"project_id": self.project.pk, "batch": value}
+        )
+        resp = self.client.get(tasks_url)
+        self.assertEqual(resp.status_code, expected_status)
+        return resp.json()
+
+    def test_tasks_endpoint_with_batch(self):
+        """Test the tasks endpoint respects the batch parameter."""
+        data = self._task_batch_helper(5, 200)
+        self.assertIn("tasks", data)
+        self.assertEqual(len(data["tasks"]), 5)
+
+    def test_tasks_endpoint_with_invalid_batch(self):
+        """Test the tasks endpoint with bad batch parameters."""
+
+        for value in ["invalid", None, "", 0]:
+            with self.subTest(batch=value):
+                self._task_batch_helper(value, 400)
+
+    def test_tasks_endpoint_without_pipeline(self):
+        """Test the tasks endpoint returns error when job has no pipeline."""
+        # Use the existing job which doesn't have a pipeline
+        job_data = self._create_job("Job without pipeline", start_now=False)
+
+        self.client.force_authenticate(user=self.user)
+        tasks_url = reverse_with_params(
+            "api:job-tasks", args=[job_data["id"]], params={"project_id": self.project.pk, "batch": 1}
+        )
+        resp = self.client.get(tasks_url)
+
+        self.assertEqual(resp.status_code, 400)
+        self.assertIn("pipeline", resp.json()[0].lower())
+
+    def test_result_endpoint_stub(self):
+        """Test the result endpoint accepts results (stubbed implementation)."""
+        pipeline = self._create_pipeline()
+        job = self._create_ml_job("Job for results test", pipeline)
+
+        self.client.force_authenticate(user=self.user)
+        result_url = reverse_with_params(
+            "api:job-result", args=[job.pk], params={"project_id": self.project.pk, "batch": 1}
+        )
+
+        result_data = [
+            {
+                "reply_subject": "test.reply.1",
+                "result": {
+                    "pipeline": "test-pipeline",
+                    "algorithms": {},
+                    "total_time": 1.5,
+                    "source_images": [],
+                    "detections": [],
+                    "errors": None,
+                },
+            }
+        ]
+
+        resp = self.client.post(result_url, result_data, format="json")
+
+        self.assertEqual(resp.status_code, 200)
+        data = resp.json()
+        self.assertEqual(data["status"], "received")
+        self.assertEqual(data["job_id"], job.pk)
+        self.assertEqual(data["results_received"], 1)
+        self.assertIn("message", data)
+
+    def test_result_endpoint_validation(self):
+        """Test the result endpoint validates request data."""
+        pipeline = self._create_pipeline()
+        job = self._create_ml_job("Job for validation test", pipeline)
+
+        self.client.force_authenticate(user=self.user)
+        result_url = reverse_with_params("api:job-result", args=[job.pk], params={"project_id": self.project.pk})
+
+        # Test with missing reply_subject
+        invalid_data = [{"result": {"pipeline": "test"}}]
+        resp = self.client.post(result_url, invalid_data, format="json")
+        self.assertEqual(resp.status_code, 400)
+        self.assertIn("reply_subject", resp.json()[0].lower())
+
+        # Test with missing result
+        invalid_data = [{"reply_subject": "test.reply"}]
+        resp = self.client.post(result_url, invalid_data, format="json")
+        self.assertEqual(resp.status_code, 400)
+        self.assertIn("result", resp.json()[0].lower())