Merge pull request #13 from Amsterdam/123195/feature-bpmn-view

123195/feature bpmn view

.flake8

@@ -0,0 +1,5 @@
+[flake8]
+ignore = E203, E266, E501, W503, F403, F401, E231
+max-line-length = 79
+max-complexity = 18
+select = B,C,E,F,W,T4,B9

app/apps/cases/admin.py

@@ -1,9 +1,6 @@
 from django.contrib import admin
-
 from .models import Case
 
-# Register your models here.
-
 
 @admin.register(Case)
 class CaseAdmin(admin.ModelAdmin):

app/apps/workflow/serializers.py

@@ -144,3 +144,13 @@ def run_validation(self, data=empty):
                 raise ValueError
 
         return super().run_validation(data)
+
+
+class BpmnModelListSerializer(serializers.ListSerializer):
+    child = serializers.CharField(max_length=100)
+
+
+class BpmnModelSerializer(serializers.Serializer):
+    version = serializers.CharField(max_length=100)
+    file_name = serializers.CharField(max_length=100)
+    model = serializers.CharField(max_length=100)

app/apps/workflow/utils.py

@@ -1,9 +1,81 @@
 import datetime
+import glob
 import json
-
+import os
 from django.conf import settings
 
 
+def get_bpmn_models():
+    path = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)), "bpmn_files", "default"
+    )
+    # Ensure the path exists
+    if not os.path.isdir(path):
+        return []
+
+    # Get all subdirectories in the specified path
+    try:
+        dirs = [d for d in os.listdir(path) if os.path.isdir(os.path.join(path, d))]
+        dirs.sort()
+        return dirs
+    except Exception as e:
+        # Handle exceptions (e.g., permission issues)
+        return {"error": str(e)}
+
+
+def get_bpmn_model_versions_and_files(model_name):
+    base_path = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)), "bpmn_files", "default", model_name
+    )
+
+    if not os.path.isdir(base_path):
+        return {"error": f"Model '{model_name}' not found"}, 404
+
+    versions = []
+
+    for version_dir in os.listdir(base_path):
+        version_path = os.path.join(base_path, version_dir)
+        if os.path.isdir(version_path):
+            # Assuming there's only one .bpmn file per version
+            bpmn_files = glob.glob(os.path.join(version_path, "*.bpmn"))
+            if bpmn_files:
+                version = version_dir
+                bpmn_file_name = os.path.basename(
+                    bpmn_files[0]
+                )  # Get the first .bpmn file
+                versions.append(
+                    {
+                        "version": version,
+                        "file_name": bpmn_file_name,
+                        "model": model_name,
+                    }
+                )
+
+    # Sort versions by the 'version' key. Assuming versions are semver or can be lexicographically sorted
+    versions.sort(key=lambda x: tuple(map(int, x["version"].split("."))))
+
+    return versions
+
+
+def get_bpmn_file(model_name, version):
+    # Construct the file path based on the provided parameters
+    path = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)),
+        "bpmn_files",
+        "default",
+        model_name,
+        version,
+        f"{model_name}.bpmn",
+    )
+
+    # Check if the file exists
+    if os.path.exists(path):
+        # Read the file content
+        with open(path, "r", encoding="utf-8") as file:
+            content = file.read()
+            return content
+
+
 def map_variables_on_task_spec_form(variables, task_spec_form):
     # transforms form result data and adds labels for the frontend
     form = dict((f.get("name"), f) for f in task_spec_form)

app/apps/workflow/views.py

@@ -1,17 +1,27 @@
-from apps.workflow.utils import map_variables_on_task_spec_form
+import logging
+from apps.workflow.utils import (
+    get_bpmn_models,
+    get_bpmn_file,
+    get_bpmn_model_versions_and_files,
+    map_variables_on_task_spec_form,
+)
 from django.http import HttpResponse, HttpResponseBadRequest
 from drf_spectacular.utils import extend_schema
-from rest_framework import viewsets
-from rest_framework.decorators import action
 from rest_framework import mixins, viewsets
+from rest_framework.response import Response
+from rest_framework.decorators import action
 
 from .models import CaseUserTask, GenericCompletedTask
 from .serializers import (
+    BpmnModelListSerializer,
+    BpmnModelSerializer,
     CaseUserTaskSerializer,
     GenericCompletedTaskCreateSerializer,
     GenericCompletedTaskSerializer,
 )
 
+logger = logging.getLogger(__name__)
+
 
 class CaseUserTaskViewSet(
     viewsets.GenericViewSet,
@@ -75,3 +85,51 @@ def complete_task(self, request):
             except Exception as e:
                 raise e
         return HttpResponseBadRequest("Invalid request")
+
+
+class BpmnViewSet(viewsets.GenericViewSet):
+    @extend_schema(
+        description="Get all BPMN model names",
+        responses={200: BpmnModelListSerializer},  # Array of strings
+    )
+    def list(self, request):
+        try:
+            models = get_bpmn_models()  # Returns a list of model names
+            return Response(models, status=200)
+        except Exception as e:
+            logger.error(f"Failed to fetch BPMN model names: {e}")
+            return Response(
+                {"detail": "An error occurred while fetching BPMN model names"},
+                status=500,
+            )
+
+    @extend_schema(
+        description="Get versions and filenames for a specific model",
+        responses={200: BpmnModelSerializer(many=True)},
+    )
+    @action(detail=False, url_path="(?P<model_name>[^/]+)", methods=["get"])
+    def get_model_versions(self, request, model_name):
+        try:
+            versions = get_bpmn_model_versions_and_files(model_name)
+            if isinstance(versions, dict) and "error" in versions:
+                return Response(versions, status=404)
+            return Response(versions, status=200)
+        except Exception as e:
+            logger.error(f"Failed to fetch BPMN models: {e}")
+            return Response(
+                {"detail": "An error occurred while fetching BPMN models."}, status=500
+            )
+
+    @extend_schema(
+        description="Get a specific BPMN workflow file",
+        responses={200: None},
+    )
+    @action(
+        detail=False,
+        url_path="(?P<model_name>[^/]+)/file/(?P<version>[^/]+)",
+        methods=["get"],
+    )
+    def get_bpmn_file(self, request, model_name, version):
+        content = get_bpmn_file(model_name, version)
+        # Return the file content as an XML response
+        return HttpResponse(content, content_type="application/xml")

app/config/urls.py

@@ -1,5 +1,9 @@
 from apps.cases.views import CaseViewSet
-from apps.workflow.views import CaseUserTaskViewSet, GenericCompletedTaskViewSet
+from apps.workflow.views import (
+    CaseUserTaskViewSet,
+    GenericCompletedTaskViewSet,
+    BpmnViewSet,
+)
 from django.conf.urls import include
 from django.contrib import admin
 from django.http import HttpResponse
@@ -11,6 +15,7 @@
 router.register(r"cases", CaseViewSet, basename="cases")
 router.register(r"generic-tasks", GenericCompletedTaskViewSet, basename="generictasks")
 router.register(r"tasks", CaseUserTaskViewSet, basename="tasks")
+router.register(r"bpmn-models", BpmnViewSet, basename="bpmn-models")
 
 
 def ok(request):