Merge pull request #378 from jbernal0019/master

Implement the workflows Django app

Author: jbernal0019

.github/workflows/dev.yml

@@ -41,7 +41,8 @@
     'pacsfiles',
     'servicefiles',
     'filebrowser',
-    'users'
+    'users',
+    'workflows'
 ]
 
 # Pagination

chris_backend/config/settings/common.py

@@ -72,7 +72,7 @@
 
 for app in ['collectionjson', 'core', 'feeds', 'plugins', 'plugininstances', 'pipelines',
             'pipelineinstances', 'uploadedfiles', 'pacsfiles', 'servicefiles', 'users',
-            'filebrowser']:
+            'filebrowser', 'workflows']:
     LOGGING['loggers'][app] = {
             'level': 'DEBUG',
             'handlers': ['console_verbose', 'file'],

chris_backend/config/settings/local.py

@@ -9,6 +9,7 @@
 from plugininstances import views as plugininstance_views
 from pipelines import views as pipeline_views
 from pipelineinstances import views as pipelineinstance_views
+from workflows import views as workflow_views
 from uploadedfiles import views as uploadedfile_views
 from pacsfiles import views as pacsfile_views
 from servicefiles import views as servicefile_views
@@ -276,6 +277,23 @@
         name='pipelineinstance-plugininstance-list'),
 
 
+    path('v1/pipelines/<int:pk>/workflows/',
+         workflow_views.WorkflowList.as_view(),
+         name='workflow-list'),
+
+    path('v1/pipelines/workflows/',
+         workflow_views.AllWorkflowList.as_view(),
+         name='allworkflow-list'),
+
+    path('v1/pipelines/workflows/search/',
+         workflow_views.AllWorkflowListQuerySearch.as_view(),
+         name='allworkflow-list-query-search'),
+
+    path('v1/pipelines/workflows/<int:pk>/',
+         workflow_views.WorkflowDetail.as_view(),
+         name='workflow-detail'),
+
+
     path('v1/uploadedfiles/',
         uploadedfile_views.UploadedFileList.as_view(),
         name='uploadedfile-list'),

chris_backend/core/api.py

@@ -303,6 +303,7 @@ def list(self, request, *args, **kwargs):
                  'pipelines': reverse('pipeline-list', request=request),
                  'pipeline_instances': reverse('allpipelineinstance-list',
                                                request=request),
+                 'workflows': reverse('allworkflow-list', request=request),
                  'tags': reverse('tag-list', request=request),
                  'uploadedfiles': reverse('uploadedfile-list', request=request),
                  'pacsfiles': reverse('pacsfile-list', request=request),

chris_backend/feeds/views.py

@@ -48,7 +48,7 @@ def validate_previous_plugin_inst(self, previous_plugin_inst_id):
         except (ValueError, ObjectDoesNotExist):
             raise serializers.ValidationError(
                 {'previous_plugin_inst_id':
-                     ["Couldn't find any 'previous' plugin instance with id %s."]})
+                     [f"Couldn't find any 'previous' plugin instance with id {pk}."]})
         # check that the user can run plugins within this feed
         user = self.context['request'].user
         if user not in previous_plugin_inst.feed.owner.all():

chris_backend/pipelineinstances/serializers.py

@@ -27,13 +27,14 @@ class PipelineSerializer(serializers.HyperlinkedModelSerializer):
     default_parameters = serializers.HyperlinkedIdentityField(
         view_name='pipeline-defaultparameter-list')
     instances = serializers.HyperlinkedIdentityField(view_name='pipelineinstance-list')
+    workflows = serializers.HyperlinkedIdentityField(view_name='workflow-list')
 
     class Meta:
         model = Pipeline
         fields = ('url', 'id', 'name', 'locked', 'authors', 'category', 'description',
                   'plugin_tree', 'plugin_inst_id', 'owner_username', 'creation_date',
                   'modification_date', 'plugins', 'plugin_pipings', 'default_parameters',
-                  'instances')
+                  'instances', 'workflows')
 
     def create(self, validated_data):
         """
@@ -118,10 +119,10 @@ def validate_plugin_inst_id(self, plugin_inst_id):
             raise serializers.ValidationError(
                 ["Couldn't find any plugin instance with id %s." % plugin_inst_id])
         plg = plg_inst.plugin
-        if plg.meta.type == 'fs':
+        if plg.meta.type != 'ds':
             raise serializers.ValidationError(
-                ["Plugin instance of %s which is of type 'fs' and therefore can not "
-                 "be used as the root of a new pipeline." % plg.meta.name])
+                [f"Plugin instance of %s which is of type {plg.meta.type} and therefore "
+                 f"can not be used as the root of a new pipeline." % plg.meta.name])
         return plg_inst
 
     def validate_plugin_tree(self, plugin_tree):
@@ -313,19 +314,21 @@ def _add_plugin_tree_to_pipeline(pipeline, tree_dict):
 
 class PluginPipingSerializer(serializers.HyperlinkedModelSerializer):
     plugin_id = serializers.ReadOnlyField(source='plugin.id')
+    plugin_name = serializers.ReadOnlyField(source='plugin.meta.name')
+    plugin_version = serializers.ReadOnlyField(source='plugin.version')
     pipeline_id = serializers.ReadOnlyField(source='pipeline.id')
     previous_id = serializers.ReadOnlyField(source='previous.id')
     previous = serializers.HyperlinkedRelatedField(view_name='pluginpiping-detail',
-                                                 read_only=True)
+                                                   read_only=True)
     plugin = serializers.HyperlinkedRelatedField(view_name='plugin-detail',
                                                  read_only=True)
     pipeline = serializers.HyperlinkedRelatedField(view_name='pipeline-detail',
                                                    read_only=True)
 
     class Meta:
         model = PluginPiping
-        fields = ('url', 'id', 'plugin_id', 'pipeline_id', 'previous_id', 'previous',
-                  'plugin', 'pipeline')
+        fields = ('url', 'id', 'previous_id', 'plugin_id', 'plugin_name',
+                  'plugin_version', 'pipeline_id', 'previous', 'plugin', 'pipeline')
 
 
 class DefaultPipingStrParameterSerializer(serializers.HyperlinkedModelSerializer):

chris_backend/pipelines/serializers.py

@@ -353,3 +353,11 @@ class Meta:
 
     def __str__(self):
         return self.value
+
+
+PARAMETER_MODELS = {'string': StrParameter,
+                    'integer': IntParameter,
+                    'float': FloatParameter,
+                    'boolean': BoolParameter,
+                    'path': PathParameter,
+                    'unextpath': UnextpathParameter}

chris_backend/plugininstances/models.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

chris_backend/workflows/__init__.py

@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class WorkflowsConfig(AppConfig):
+    name = 'workflows'

chris_backend/workflows/admin.py

@@ -0,0 +1,31 @@
+# Generated by Django 2.2.24 on 2022-02-27 00:49
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('pipelines', '0005_auto_20200213_0040'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Workflow',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('creation_date', models.DateTimeField(auto_now_add=True)),
+                ('created_plugin_inst_ids', models.CharField(max_length=600)),
+                ('owner', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+                ('pipeline', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='workflows', to='pipelines.Pipeline')),
+            ],
+            options={
+                'ordering': ('-creation_date',),
+            },
+        ),
+    ]

chris_backend/workflows/apps.py

@@ -0,0 +1,31 @@
+
+from django.db import models
+import django_filters
+from django_filters.rest_framework import FilterSet
+
+from pipelines.models import Pipeline
+
+
+class Workflow(models.Model):
+    creation_date = models.DateTimeField(auto_now_add=True)
+    created_plugin_inst_ids = models.CharField(max_length=600)
+    pipeline = models.ForeignKey(Pipeline, on_delete=models.CASCADE,
+                                 related_name='workflows')
+    owner = models.ForeignKey('auth.User', on_delete=models.CASCADE)
+
+    class Meta:
+        ordering = ('-creation_date',)
+
+    def __str__(self):
+        return self.created_plugin_inst_ids
+
+
+class WorkflowFilter(FilterSet):
+    pipeline_name = django_filters.CharFilter(field_name='pipeline__name',
+                                               lookup_expr='icontains')
+    owner_username = django_filters.CharFilter(field_name='owner__username',
+                                               lookup_expr='exact')
+
+    class Meta:
+        model = Workflow
+        fields = ['id', 'pipeline_name', 'owner_username']

chris_backend/workflows/migrations/0001_initial.py

@@ -0,0 +1,17 @@
+from rest_framework import permissions
+
+
+class IsOwnerOrChrisOrReadOnly(permissions.BasePermission):
+    """
+    Custom permission to only allow owners of an object or superuser
+    'chris' to modify/edit it. Read only is allowed to other users.
+    """
+
+    def has_object_permission(self, request, view, obj):
+        # Read permissions are allowed to any request,
+        # so we'll always allow GET, HEAD or OPTIONS requests.
+        if request.method in permissions.SAFE_METHODS:
+            return True
+
+        # Write permissions are only allowed to the owner and superuser 'chris'.
+        return (request.user == obj.owner) or (request.user.username == 'chris')

chris_backend/workflows/migrations/__init__.py

@@ -0,0 +1,144 @@
+
+import json
+
+from django.core.exceptions import ObjectDoesNotExist
+from rest_framework import serializers
+
+from plugininstances.models import PluginInstance
+from plugininstances.serializers import PluginInstanceSerializer
+from pipelines.serializers import DEFAULT_PIPING_PARAMETER_SERIALIZERS
+
+from .models import Workflow
+
+
+class WorkflowSerializer(serializers.HyperlinkedModelSerializer):
+    created_plugin_inst_ids = serializers.ReadOnlyField()
+    pipeline_id = serializers.ReadOnlyField(source='pipeline.id')
+    pipeline_name = serializers.ReadOnlyField(source='pipeline.name')
+    previous_plugin_inst_id = serializers.IntegerField(min_value=1, write_only=True)
+    nodes_info = serializers.JSONField(write_only=True)
+    owner_username = serializers.ReadOnlyField(source='owner.username')
+    pipeline = serializers.HyperlinkedRelatedField(view_name='pipeline-detail',
+                                                   read_only=True)
+
+    class Meta:
+        model = Workflow
+        fields = ('url', 'id', 'creation_date', 'created_plugin_inst_ids', 'pipeline_id',
+                  'pipeline_name', 'owner_username', 'previous_plugin_inst_id',
+                  'nodes_info', 'pipeline')
+
+    def create(self, validated_data):
+        """
+        Overriden to delete 'previous_plugin_inst_id' and 'nodes_info' from serializer
+        data as they are not model fields.
+        """
+        del validated_data['previous_plugin_inst_id']
+        del validated_data['nodes_info']
+        return super(WorkflowSerializer, self).create(validated_data)
+
+    def validate_previous_plugin_inst_id(self, previous_plugin_inst_id):
+        """
+        Overriden to check that an integer id is provided for previous plugin instance.
+        Then check that the id exists in the DB and that the user can run plugins
+        within the corresponding feed.
+        """
+        if not previous_plugin_inst_id:
+            raise serializers.ValidationError(['This field is required.'])
+        try:
+            pk = int(previous_plugin_inst_id)
+            previous_plugin_inst = PluginInstance.objects.get(pk=pk)
+        except (ValueError, ObjectDoesNotExist):
+            raise serializers.ValidationError(
+                [f"Couldn't find any 'previous' plugin instance with id "
+                 f"{previous_plugin_inst_id}."])
+        # check that the user can run plugins within this feed
+        user = self.context['request'].user
+        if user not in previous_plugin_inst.feed.owner.all():
+            raise serializers.ValidationError([f'User is not an owner of feed for '
+                                               f'previous instance with id {pk}.'])
+        return previous_plugin_inst
+
+    def validate_nodes_info(self, nodes_info):
+        """
+        Overriden to validate the runtime data for the workflow. It should be a
+        JSON string encoding a list of dictionaries. Each dictionary is a workflow node
+        containing a plugin piping_id, compute_resource_name, title and a list of
+        dictionaries called plugin_parameter_defaults. Each dictionary in this list has
+        name and default keys.
+        """
+        try:
+            node_list = list(json.loads(nodes_info))
+        except json.decoder.JSONDecodeError:
+            # overriden validation methods automatically add the field name to the msg
+            raise serializers.ValidationError([f'Invalid JSON string {nodes_info}.'])
+        except Exception:
+            raise serializers.ValidationError([f'Invalid list in {nodes_info}'])
+
+        pipeline = self.context['view'].get_object()
+        pipings = list(pipeline.plugin_pipings.all())
+        if len(node_list) != len(pipings):
+            raise serializers.ValidationError(
+                [f'Invalid length for list in {nodes_info}'])
+
+        for piping in pipings:
+            d_l = [d for d in node_list if d.get('piping_id') == piping.id]
+            try:
+                d = d_l[0]
+            except IndexError:
+                raise serializers.ValidationError([f'Missing data for plugin pipping '
+                                                   f'with id {piping.id}'])
+            cr_name = d.get('compute_resource_name')
+            if not cr_name:
+                raise serializers.ValidationError([f'Missing compute_resource_name key in'
+                                                   f' {d}'])
+            if piping.plugin.compute_resources.filter(name=cr_name).count() == 0:
+                msg = [f'Plugin for pipping with id {piping.id} has not been registered '
+                       f'with a compute resource named {cr_name}']
+                raise serializers.ValidationError(msg)
+
+            title = d.get('title', '')
+            plg_inst_serializer = PluginInstanceSerializer(data={'title': title})
+            try:
+                plg_inst_serializer.is_valid(raise_exception=True)
+            except Exception:
+                msg = [f'Invalid title {title} for pipping with id {piping.id}']
+                raise serializers.ValidationError(msg)
+
+            piping_param_defaults = d.get('plugin_parameter_defaults', [])
+            self.validate_piping_params(piping.id, piping.string_param.all(),
+                                        piping_param_defaults)
+            self.validate_piping_params(piping.id, piping.integer_param.all(),
+                                        piping_param_defaults)
+            self.validate_piping_params(piping.id, piping.float_param.all(),
+                                        piping_param_defaults)
+            self.validate_piping_params(piping.id, piping.boolean_param.all(),
+                                        piping_param_defaults)
+        return node_list
+
+    @staticmethod
+    def validate_piping_params(piping_id, piping_default_params, piping_param_defaults):
+        """
+        Helper method to validate that if a default value doesn't exist in the
+        corresponding pipeline for a piping parameter then a default is provided for it
+        when creating a new runtime workflow.
+        """
+        for default_param in piping_default_params:
+            l = [d for d in piping_param_defaults if
+                 d.get('name') == default_param.plugin_param.name]
+            if default_param.value is None and (not l or l[0].get('default') is None):
+                raise serializers.ValidationError(
+                    [f"Can not run workflow. Parameter "
+                     f"'{default_param.plugin_param.name}' for piping with id "
+                     f"{piping_id} does not have a default value in the pipeline"])
+            if l and l[0].get('default'):
+                param_default = l[0].get('default')
+                param_type = default_param.plugin_param.type
+                default_serializer_cls = DEFAULT_PIPING_PARAMETER_SERIALIZERS[param_type]
+                default_serializer = default_serializer_cls(data={'value': param_default})
+                try:
+                    default_serializer.is_valid(raise_exception=True)
+                except Exception:
+                    msg = [f'Invalid parameter default value {param_default} for '
+                           f"parameter '{default_param.plugin_param.name}' and pipping "
+                           f'with id {piping_id}']
+                    raise serializers.ValidationError(msg)