Add create method on serializer

Author: koopmant

app/grandchallenge/algorithms/models.py

@@ -1762,7 +1762,7 @@ class InvocationStatusChoices(TextChoices):
     CANCELLED = "CANCELLED", _("Cancelled")
 
 
-class Invocation(UUIDModel):
+class Invocation(CIVForObjectMixin, UUIDModel):
     StatusChoices = InvocationStatusChoices
 
     endpoint = models.ForeignKey(Endpoint, on_delete=models.PROTECT)
@@ -1861,6 +1861,13 @@ def orchestrator_kwargs(self):
     def orchestrator(self):
         return EndpointOrchestrator(**self.orchestrator_kwargs)
 
+    @property
+    def is_editable(self):
+        if self.status == InvocationStatusChoices.VALIDATING_INPUTS:
+            return True
+        else:
+            return False
+
     @cached_property
     def inputs_complete(self):
         # check if all inputs are present and if they all have a value
@@ -1894,3 +1901,31 @@ def update_status(
             self.invoke_duration = invoke_duration
 
         self.save()
+
+    def add_civ(self, *, civ):
+        super().add_civ(civ=civ)
+        return self.inputs.add(civ)
+
+    def remove_civ(self, *, civ):
+        super().remove_civ(civ=civ)
+        return self.inputs.remove(civ)
+
+    def get_civ_for_interface(self, interface):
+        return self.inputs.get(interface=interface)
+
+    def process_civ_data_objects_and_execute_linked_task(
+        self, *, civ_data_objects, user, linked_task=None
+    ):
+        from grandchallenge.algorithms.tasks import (
+            execute_invocation_for_inputs,
+        )
+
+        linked_task = execute_invocation_for_inputs.signature(
+            kwargs={"invocation_pk": str(self.pk)}, immutable=True
+        )
+
+        return super().process_civ_data_objects_and_execute_linked_task(
+            civ_data_objects=civ_data_objects,
+            user=user,
+            linked_task=linked_task,
+        )

app/grandchallenge/algorithms/serializers.py

@@ -21,6 +21,7 @@
     AlgorithmModel,
     Endpoint,
     Invocation,
+    InvocationStatusChoices,
     Job,
     annotate_input_output_counts,
 )
@@ -399,3 +400,31 @@ def validate(self, data):
         )
 
         return data
+
+    def create(self, validated_data):
+        civ_data_objects = validated_data.pop("civ_data_objects", [])
+
+        invocation = Invocation.objects.create(
+            **validated_data,
+            status=InvocationStatusChoices.VALIDATING_INPUTS,
+        )
+
+        try:
+            invocation.process_civ_data_objects_and_execute_linked_task(
+                civ_data_objects=civ_data_objects,
+                user=self.context["request"].user,
+            )
+        except CIVNotEditableException as e:
+            invocation.refresh_from_db()
+            if invocation.status == invocation.StatusChoices.CANCELLED:
+                # this can happen for jobs with multiple inputs
+                # if one of them fails validation
+                pass
+            else:
+                error_handler = invocation.get_error_handler()
+                error_handler.handle_error(
+                    error_message=SystemErrorMessages.UNEXPECTED_ERROR,
+                )
+                logger.error(e, exc_info=True)
+
+        return invocation

app/grandchallenge/algorithms/tasks.py

@@ -12,6 +12,7 @@
 from grandchallenge.algorithms.exceptions import TooManyJobsScheduled
 from grandchallenge.components.schemas import GPUTypeChoices
 from grandchallenge.components.tasks import (
+    provision_invocation_input_data,
     remove_container_image_from_registry,
 )
 from grandchallenge.core.celery import (
@@ -325,3 +326,31 @@ def deactivate_old_algorithm_images():
                 }
             ).apply_async
         )
+
+
+@acks_late_micro_short_task(retry_on=(LockNotAcquiredException,))
+@transaction.atomic
+def execute_invocation_for_inputs(*, invocation_pk):
+    from grandchallenge.algorithms.models import Invocation
+
+    with check_lock_acquired():
+        invocation = Invocation.objects.select_for_update(nowait=True).get(
+            pk=invocation_pk
+        )
+
+    if not invocation.inputs_complete:
+        # Nothing to do
+        return
+
+    if invocation.status != Invocation.StatusChoices.VALIDATING_INPUTS:
+        # this task can be called multiple times with complete inputs,
+        # and might have been queued for execution already, so ignore
+        return
+
+    invocation.update_status(status=Invocation.StatusChoices.QUEUED)
+
+    on_commit(
+        provision_invocation_input_data.signature(
+            kwargs=invocation.task_kwargs
+        ).apply_async
+    )

app/grandchallenge/components/models.py

@@ -73,6 +73,7 @@
     DICOMImageSetUploadErrorHandler,
     EvaluationCIVErrorHandler,
     FallbackCIVValidationErrorHandler,
+    InvocationCIVErrorHandler,
     JobCIVErrorHandler,
     RawImageUploadSessionErrorHandler,
     UserUploadCIVErrorHandler,
@@ -2841,7 +2842,7 @@ def get_current_value_for_interface(self, *, interface, user):
 
     def get_error_handler(self, *, linked_object=None):
         # local imports to prevent circular dependency
-        from grandchallenge.algorithms.models import Job
+        from grandchallenge.algorithms.models import Invocation, Job
         from grandchallenge.archives.models import ArchiveItem
         from grandchallenge.evaluation.models import Evaluation
         from grandchallenge.reader_studies.models import DisplaySet
@@ -2860,6 +2861,8 @@ def get_error_handler(self, *, linked_object=None):
             return JobCIVErrorHandler(job=self)
         elif isinstance(self, Evaluation):
             return EvaluationCIVErrorHandler(job=self)
+        elif isinstance(self, Invocation):
+            return InvocationCIVErrorHandler(invocation=self)
         elif linked_object and isinstance(linked_object, UserUpload):
             return UserUploadCIVErrorHandler(
                 user_upload=linked_object,

app/grandchallenge/core/error_handlers.py

@@ -86,6 +86,41 @@ def handle_error(self, *, error_message, interface=None, user=None):
         )
 
 
+class InvocationCIVErrorHandler(ErrorHandler):
+    """
+    Error handler for CIV validation errors on invocation creation.
+    Handle_error() updates an algorithm endpoint invocation.
+    """
+
+    def __init__(self, *args, invocation, **kwargs):
+        from grandchallenge.algorithms.models import Invocation
+
+        if not invocation or not isinstance(invocation, Invocation):
+            raise RuntimeError(
+                "You need to provide an Invocation instance to this error handler."
+            )
+
+        super().__init__(*args, **kwargs)
+        self._invocation = invocation
+
+    def handle_error(self, *, error_message, interface=None, user=None):
+        if interface:
+            detailed_error_message = copy.deepcopy(
+                self._invocation.detailed_error_message
+            )
+            detailed_error_message[interface.title] = error_message
+            self._invocation.update_status(
+                status=self._invocation.StatusChoices.CANCELLED,
+                error_message="One or more of the inputs failed validation.",
+                detailed_error_message=detailed_error_message,
+            )
+        else:
+            self._invocation.update_status(
+                status=self._invocation.StatusChoices.CANCELLED,
+                error_message=error_message,
+            )
+
+
 class RawImageUploadSessionErrorHandler(ErrorHandler):
     """
     Error handler for image imports and image validation.

app/tests/algorithms_tests/test_serializers.py

@@ -758,24 +758,57 @@ def test_time_limit_validation_on_invocation_post_serializer(settings):
 
 
 @pytest.mark.django_db
-def test_invocation_post_serializer_create(request):
-    endpoint = EndpointFactory(status=Endpoint.StatusChoices.RUNNING)
-    request.user = endpoint.creator
-    socket = ComponentInterfaceFactory(kind=ComponentInterface.Kind.STRING)
-    interface = AlgorithmInterfaceFactory(inputs=[socket])
+def test_invocation_post_serializer_create(
+    request, settings, django_capture_on_commit_callbacks
+):
+    settings.CELERY_TASK_ALWAYS_EAGER = True
+    settings.CELERY_TASK_EAGER_PROPAGATES = True
+
+    user = UserFactory()
+    request.user = user
+    endpoint = EndpointFactory.create(
+        creator=user,
+        status=Endpoint.StatusChoices.RUNNING,
+    )
+    ci_string = ComponentInterfaceFactory.create(
+        kind=ComponentInterface.Kind.STRING
+    )
+    ci_img1, ci_img2 = ComponentInterfaceFactory.create_batch(
+        2, kind=ComponentInterface.Kind.PANIMG_IMAGE
+    )
+    interface = AlgorithmInterfaceFactory(inputs=[ci_string, ci_img2, ci_img1])
     endpoint.algorithm_image.algorithm.interfaces.add(interface)
+    upload = RawImageUploadSessionFactory(creator=user)
+    image1, image2 = ImageFactory.create_batch(2)
+    for im in [image1, image2]:
+        assign_perm("view_image", user, im)
+    upload.image_set.set([image1])
 
     serializer = InvocationPostSerializer(
         data={
             "endpoint": endpoint.api_url,
-            "inputs": [{"interface": socket.slug, "value": "dummy"}],
+            "inputs": [
+                {"interface": ci_string.slug, "value": "foo"},
+                {"interface": ci_img1.slug, "upload_session": upload.api_url},
+                {"interface": ci_img2.slug, "image": image2.api_url},
+            ],
         },
         context={"request": request},
     )
 
-    assert serializer.is_valid()
-    serializer.create(serializer.validated_data)
+    assert serializer.is_valid(), serializer.errors
+
+    # fake successful upload
+    upload.status = RawImageUploadSession.SUCCESS
+    upload.save()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        serializer.create(serializer.validated_data)
+
+    assert Invocation.objects.count() == 1
 
     invocation = Invocation.objects.get()
 
     assert invocation.endpoint == endpoint
+    assert invocation.algorithm_interface == interface
+    assert invocation.inputs.count() == 3