add migration example

spiff_example/cli/diff_result.py

@@ -0,0 +1,68 @@
+import sys
+
+from SpiffWorkflow import TaskState
+
+def get_name_and_start(spec_diff):
+    for ts in spec_diff.alignment:
+        if ts == ts._wf_spec.start:
+            return ts._wf_spec.name, ts
+
+def show_spec_diff(spec_diff):
+
+    name, start = get_name_and_start(spec_diff)
+    updates, checked = [], []
+
+    # This determines what needs to be be displayed (omits specs with exact mappings)
+    def order_specs(ts):
+        for output in [s for s in ts.outputs if s not in checked]:
+            checked.append(output)
+            if output in spec_diff.changed or spec_diff.alignment[output] is None:
+                updates.append(output)
+            order_specs(output)
+
+    order_specs(start)
+
+    sys.stdout.write(f'\nResults for {name}\n\n')
+    sys.stdout.write('Changed or Removed\n')
+
+    for ts in updates:
+        # Ignore tasks that don't appear in diagrams
+        detail = ''
+        if ts.bpmn_id is not None:
+            if spec_diff.alignment[ts] is None:
+                detail = 'removed'
+            elif ts in spec_diff.changed:
+                detail = ', '.join(spec_diff.changed[ts])
+        if detail:
+            label = f'{ts.bpmn_name or '-'} [{ts.name}]'
+            sys.stdout.write(f'{label:<50s} {detail}\n')
+
+    if len(spec_diff.added) > 0:
+        sys.stdout.write('\nAdded\n')
+        for ts in spec_diff.added:
+            if ts.bpmn_id is not None:
+                sys.stdout.write(f'{ts.bpmn_name or '-'} [{ts.name}]\n')
+
+
+def show_workflow_diff(wf_diff, task_id=None):
+
+    duplicates = []
+    if task_id is not None:
+        sys.stdout.write(f'\nResults for {task_id}\n\n')
+    else:
+        sys.stdout.write('Results\n\n')
+
+    sys.stdout.write('Removed\n')
+    for task in wf_diff.removed:
+        if task.task_spec not in duplicates:
+            duplicates.append(task.task_spec)
+            label = f'{task.task_spec.bpmn_name or '-'} [{task.task_spec.name}]'
+            sys.stdout.write(f'{label:<50s} {TaskState.get_name(task.state)}\n')
+
+    sys.stdout.write('Changed\n')
+    for task in wf_diff.changed:
+        if task.task_spec not in duplicates:
+            duplicates.append(task.task_spec)
+            label = f'{task.task_spec.bpmn_name or '-'} [{task.task_spec.name}]'
+            sys.stdout.write(f'{label:<50s} {TaskState.get_name(task.state)}\n')
+

spiff_example/cli/subcommands.py

@@ -2,16 +2,18 @@
 import json
 import logging
 
+from .diff_result import show_spec_diff, show_workflow_diff
+
 def configure_logging():
-    spiff_logger = logging.getLogger('spiff')
-    spiff_handler = logging.StreamHandler()
-    spiff_handler.setFormatter(logging.Formatter('%(asctime)s [%(name)s:%(levelname)s] (%(workflow_spec)s:%(task_spec)s) %(message)s'))
-    spiff_logger.addHandler(spiff_handler)
+    task_logger = logging.getLogger('spiff.task')
+    task_handler = logging.StreamHandler()
+    task_handler.setFormatter(logging.Formatter('%(asctime)s [%(name)s:%(levelname)s] (%(workflow_spec)s:%(task_spec)s) %(message)s'))
+    task_logger.addHandler(task_handler)
 
-    metrics_logger = logging.getLogger('spiff.metrics')
-    metrics_handler = logging.StreamHandler()
-    metrics_handler.setFormatter('%(asctime)s [%(name)s:%(levelname)s] (%(workflow_spec)s:%(task_spec)s) %(elasped)s')
-    metrics_logger.addHandler(metrics_handler)
+    wf_logger = logging.getLogger('spiff.workflow')
+    wf_handler = logging.StreamHandler()
+    wf_handler.setFormatter(logging.Formatter('%(asctime)s [%(name)s:%(levelname)s] (%(workflow_spec)s) %(message)s'))
+    wf_logger.addHandler(wf_handler)
 
 def add(engine, args):
     if args.process is not None:
@@ -23,6 +25,10 @@ def show_library(engine, args):
     for spec_id, name, filename in engine.list_specs():
         sys.stdout.write(f'{spec_id}  {name:<20s} {filename}\n')
 
+def show_workflows(engine, args):
+    for wf_id, name, filename, active, started, updated in engine.list_workflows():
+        sys.stdout.write(f'{wf_id}  {name:<20s} {active} {started} {updated or ''}\n')
+
 def run(engine, args):
     instance = engine.start_workflow(args.spec_id)
     instance.run_until_user_input_required()
@@ -32,6 +38,25 @@ def run(engine, args):
     sys.stdout.write(json.dumps(instance.data, indent=2, separators=[', ', ': ']))
     sys.stdout.write('\n')
 
+def diff_spec(engine, args):
+    spec_diff = engine.diff_spec(args.original, args.new)
+    show_spec_diff(spec_diff)
+    if args.deps:
+        diffs, added = engine.diff_dependencies(args.original, args.new)
+        for diff in diffs.values():
+            show_spec_diff(diff)
+        if len(added) > 0:
+            sys.stdout.write('\nNew subprocesses\n')
+            sys.stdout.write('\n'.join(added))
+
+def diff_workflow(engine, args):
+    wf_diff, sp_diffs = engine.diff_workflow(args.wf_id, args.spec_id)
+    show_workflow_diff(wf_diff)
+    for task_id, result in sp_diffs.items():
+        show_workflow_diff(result, task_id=task_id)
+
+def migrate(engine, args):
+    engine.migrate_workflow(args.wf_id, args.spec_id, validate=not args.force)
 
 def add_subparsers(subparsers):
 
@@ -43,12 +68,32 @@ def add_subparsers(subparsers):
     add_spec.add_argument('-d', '--dmn', dest='dmn', nargs='*', metavar='FILE', help='DMN files to load')
     add_spec.set_defaults(func=add)
 
-    list_specs = subparsers.add_parser('list', help='List available specs')
+    list_specs = subparsers.add_parser('list_specs', help='List available specs')
     list_specs.set_defaults(func=show_library)
 
+    list_wf = subparsers.add_parser('list_instances', help='List instances')
+    list_wf.set_defaults(func=show_workflows)
+
     run_wf = subparsers.add_parser('run', help='Run a workflow')
     run_wf.add_argument('-s', '--spec-id', dest='spec_id', metavar='SPEC ID', help='The ID of the spec to run')
-    run_wf.add_argument('-a', '--active-ok', dest='active', action='store_true', help='Suppress exception if the workflow does not complete')
+    run_wf.add_argument('-a', '--active-ok', dest='active', action='store_true',
+                        help='Suppress exception if the workflow does not complete')
     run_wf.set_defaults(func=run)
 
+    compare_spec = subparsers.add_parser('diff_spec', help='Compare two workflow specs')
+    compare_spec.add_argument('-o', '--original', dest='original', metavar='SPEC ID', help='The ID of the original spec')
+    compare_spec.add_argument('-n', '--new', dest='new', metavar='SPEC ID', help='The ID of the new spec')
+    compare_spec.add_argument('-d', '--include-dependencies', action='store_true', dest='deps',
+                              help='Include dependencies in the output')
+    compare_spec.set_defaults(func=diff_spec)
+
+    compare_wf = subparsers.add_parser('diff_workflow', help='Compare a workflow against a new spec')
+    compare_wf.add_argument('-w', '--wf-id', dest='wf_id', metavar='WORKFLOW ID', help='The ID of the workflow')
+    compare_wf.add_argument('-s', '--spec-id', dest='spec_id', metavar='SPEC ID', help='The ID of the new spec')
+    compare_wf.set_defaults(func=diff_workflow)
 
+    migrate_wf = subparsers.add_parser('migrate', help='Update a workflow spec')
+    migrate_wf.add_argument('-w', '--wf-id', dest='wf_id', metavar='WORKFLOW ID', help='The ID of the workflow')
+    migrate_wf.add_argument('-s', '--spec-id', dest='spec_id', metavar='SPEC ID', help='The ID of the new spec')
+    migrate_wf.add_argument('-f', '--force', dest='force', action='store_true', help='Omit validation')
+    migrate_wf.set_defaults(func=migrate)

spiff_example/curses_ui/log_view.py

@@ -47,8 +47,10 @@ def write(self, record):
         for line in trace:
             self.screen.addstr(line)
         dt = datetime.fromtimestamp(record.created).strftime('%Y-%m-%d %H:%M:%S.%f')
-        if record.name == 'spiff':
+        if record.name == 'spiff.task':
             message = f'{dt} [{record.name}:{record.levelname}] ({record.workflow_spec}:{record.task_spec}) {record.msg}'
+        elif record.name == 'spiff.workflow':
+            message = f'{dt} [{record.name}:{record.levelname}] ({record.workflow_spec}) {record.msg}'
         else:
             message = f'{dt} [{record.name}:{record.levelname}] {record.msg}'
         self.screen.addstr(f'\n{message}', self.styles.get(record.levelname, 0))

spiff_example/engine/engine.py

@@ -5,6 +5,13 @@
 from SpiffWorkflow.bpmn.specs.mixins.events.event_types import CatchingEvent
 from SpiffWorkflow.bpmn import BpmnWorkflow
 from SpiffWorkflow.bpmn.script_engine import PythonScriptEngine
+from SpiffWorkflow.bpmn.util.diff import (
+    SpecDiff,
+    diff_dependencies,
+    diff_workflow,
+    filter_tasks,
+    migrate_workflow,
+)
 from SpiffWorkflow import TaskState
 
 from .instance import Instance
@@ -82,3 +89,47 @@ def list_workflows(self, include_completed=False):
     def delete_workflow(self, wf_id):
         self.serializer.delete_workflow(wf_id)
         logger.info(f'Deleted workflow with id {wf_id}')
+
+    def diff_spec(self, original_id, new_id):
+        original, _ = self.serializer.get_workflow_spec(original_id, include_dependencies=False)
+        new, _ = self.serializer.get_workflow_spec(new_id, include_dependencies=False)
+        return SpecDiff(self.serializer.registry, original, new)
+
+    def diff_dependencies(self, original_id, new_id):
+        _, original = self.serializer.get_workflow_spec(original_id, include_dependencies=True)
+        _, new = self.serializer.get_workflow_spec(new_id, include_dependencies=True)
+        return diff_dependencies(self.serializer.registry, original, new)
+
+    def diff_workflow(self, wf_id, spec_id):
+        wf = self.serializer.get_workflow(wf_id)
+        spec, deps = self.serializer.get_workflow_spec(spec_id)
+        return diff_workflow(self.serializer.registry, wf, spec, deps)
+
+    def can_migrate(self, wf_diff, sp_diffs):
+
+        def safe(result):
+            mask = TaskState.COMPLETED|TaskState.STARTED
+            tasks = result.changed + result.removed
+            return len(filter_tasks(tasks, state=mask)) == 0
+
+        for diff in sp_diffs.values():
+            if diff is None or not safe(diff):
+                return False
+        return safe(wf_diff)
+
+    def migrate_workflow(self, wf_id, spec_id, validate=True):
+
+        wf = self.serializer.get_workflow(wf_id)
+        spec, deps = self.serializer.get_workflow_spec(spec_id)
+        wf_diff, sp_diffs = diff_workflow(self.serializer.registry, wf, spec, deps)
+
+        if validate and not self.can_migrate(wf_diff, sp_diffs):
+            raise Exception('Workflow is not safe to migrate!')
+
+        migrate_workflow(wf_diff, wf, spec)
+        for sp_id, sp in wf.subprocesses.items():
+            migrate_workflow(sp_diffs[sp_id], sp, deps.get(sp.spec.name))
+        wf.subprocess_specs = deps
+
+        self.serializer.delete_workflow(wf_id)
+        return self.serializer.create_workflow(wf, spec_id)

spiff_example/serializer/sqlite/serializer.py

@@ -197,6 +197,9 @@ def _list_workflows(self, cursor, include_completed):
         return cursor.fetchall()
 
     def _delete_workflow(self, cursor, wf_id):
+        cursor.execute("select descendant as 'id [uuid]' from workflow_dependency where root=?", (wf_id, ))
+        for sp_id in [row[0] for row in cursor]:
+            cursor.execute("delete from workflow where id=?", (sp_id, ))
         cursor.execute("delete from workflow where id=?", (wf_id, ))
 
     def execute(self, func, *args, **kwargs):

spiff_example/spiff/diffs.py

@@ -0,0 +1,62 @@
+import sqlite3
+import logging
+import datetime
+
+from SpiffWorkflow.spiff.parser.process import SpiffBpmnParser
+from SpiffWorkflow.spiff.specs.defaults import UserTask, ManualTask
+from SpiffWorkflow.spiff.serializer.config import SPIFF_CONFIG
+from SpiffWorkflow.bpmn.workflow import BpmnWorkflow, BpmnSubWorkflow
+from SpiffWorkflow.bpmn.specs import BpmnProcessSpec
+from SpiffWorkflow.bpmn.specs.mixins.none_task import NoneTask
+from SpiffWorkflow.bpmn.script_engine import TaskDataEnvironment
+
+from ..serializer.sqlite import (
+    SqliteSerializer,
+    WorkflowConverter,
+    SubworkflowConverter,
+    WorkflowSpecConverter
+)
+from ..engine import BpmnEngine
+from .curses_handlers import UserTaskHandler, ManualTaskHandler
+
+from .product_info import (
+    ProductInfo,
+    product_info_to_dict,
+    product_info_from_dict,
+    lookup_product_info,
+    lookup_shipping_cost,
+)
+
+logger = logging.getLogger('spiff_engine')
+logger.setLevel(logging.INFO)
+
+spiff_logger = logging.getLogger('spiff')
+spiff_logger.setLevel(logging.INFO)
+
+SPIFF_CONFIG[BpmnWorkflow] = WorkflowConverter
+SPIFF_CONFIG[BpmnSubWorkflow] = SubworkflowConverter
+SPIFF_CONFIG[BpmnProcessSpec] = WorkflowSpecConverter
+
+dbname = 'spiff.db'
+with sqlite3.connect(dbname) as db:
+    SqliteSerializer.initialize(db)
+
+registry = SqliteSerializer.configure(SPIFF_CONFIG)
+registry.register(ProductInfo, product_info_to_dict, product_info_from_dict)
+
+serializer = SqliteSerializer(dbname, registry=registry)
+
+parser = SpiffBpmnParser()
+
+handlers = {
+    UserTask: UserTaskHandler,
+    ManualTask: ManualTaskHandler,
+    NoneTask: ManualTaskHandler,
+}
+
+script_env = TaskDataEnvironment({
+    'datetime': datetime,
+    'lookup_product_info': lookup_product_info,
+    'lookup_shipping_cost': lookup_shipping_cost,
+})
+engine = BpmnEngine(parser, serializer, script_env)