From d0c925363fa4e48e8909d0c129aa2a6b0112e68d Mon Sep 17 00:00:00 2001
From: David Vogt <david.vogt@adfinis.com>
Date: Wed, 21 Apr 2021 16:23:27 +0200
Subject: [PATCH] feat(workflow): commandline tool to show workflow
configuration
This is a first attempt at visualizing a workflow.
---
.../management/commands/show_workflow.py | 80 +++++++++++++++++++
1 file changed, 80 insertions(+)
create mode 100644 caluma/caluma_workflow/management/commands/show_workflow.py
diff --git a/caluma/caluma_workflow/management/commands/show_workflow.py b/caluma/caluma_workflow/management/commands/show_workflow.py
new file mode 100644
index 000000000..d4240cc23
--- /dev/null
+++ b/caluma/caluma_workflow/management/commands/show_workflow.py
@@ -0,0 +1,80 @@
+#!/usr/bin/python
+
+
+from logging import getLogger
+
+from django.core.management.base import BaseCommand
+from django.db import transaction
+
+from caluma.caluma_workflow.models import Task, Workflow
+
+log = getLogger(__name__)
+
+
+class Command(BaseCommand):
+ """Dump workflow structure."""
+
+ help = "Show workflow structure"
+
+ def add_arguments(self, parser):
+
+ parser.add_argument(
+ dest="workflow",
+ action="store",
+ help="Slug of the workflow to show",
+ )
+
+ @transaction.atomic
+ def handle(self, *args, **options):
+ self.workflow = Workflow.objects.get(pk=options["workflow"])
+ self.visit(self.workflow)
+
+ def _print(self, indent, data):
+ ind_str = " " * indent
+ print(f"{ind_str}{data}")
+
+ def visit(self, obj, indent=0):
+ obj_type = str(type(obj).__name__)
+ method_name = f"visit_{obj_type}"
+ method = getattr(self, method_name)
+ return method(obj, indent)
+
+ def visit_Task(self, task, indent):
+ taskflows = task.task_flows.filter(workflow=self.workflow)
+ if taskflows.exists():
+ next = taskflows.get().flow.next
+ else:
+ next = None
+
+ self._print(
+ indent,
+ f"Task {task.slug} ({task.type})",
+ )
+ if task.type == Task.TYPE_COMPLETE_TASK_FORM:
+ # other tasks *may* have a form, but it's not relevant
+ self._print(indent + 1, f"Form: {task.form}")
+ self._print(indent + 1, f"Next: {next}" if next else "(last task in sequence)")
+
+ def visit_Workflow(self, workflow, indent):
+ self._print(indent, f"Workflow {workflow.slug}")
+ self._print(indent + 1, "Allowed forms:")
+ for form in workflow.allow_forms.all():
+ self.visit(form, indent + 2)
+
+ start_tasks = workflow.start_tasks.all()
+ self._print(indent + 1, "Start Tasks:")
+ start_tasks = workflow.start_tasks.all()
+ for task in start_tasks:
+ self.visit(task, indent + 2)
+ self._print(indent + 1, "Other Tasks:")
+
+ other_tasks = (
+ Task.objects.all()
+ .exclude(pk__in=start_tasks)
+ .filter(pk__in=workflow.task_flows.all().values("task"))
+ )
+ for task in other_tasks:
+ self.visit(task, indent + 2)
+
+ def visit_Form(self, form, indent):
+ self._print(indent, f"Form: {form.slug}")