Adding Target.VALIDATION to Workflows, including isTask bool (#930)

* initial cut for adding target validate

* Add documentation and example migration

* fix linting

* bump to 4.0.0rc1

* bump to 4.0.0rc2

* refactor target.system

* fix linting

* Add VALIDATE to Workflow and is_task

* fix linting

* Update helper for backwards compatibility

* Add is_task to proces from workflow table

* revert and get db workflow from make_workflow

* revert and get db workflow from make_workflow

* revert and get db workflow from make_workflow

* update readme'

* Update after review

* fix unit tests

* update linting and docs

* remove print

* remove hidden navigation

* fix linting

---------

Co-authored-by: Igor van Spengen <[email protected]>
Co-authored-by: Igor van Spengen <[email protected]>

Author: saket-geant

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.0.0rc1
+current_version = 4.0.0rc2
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(rc(?P<build>\d+))?

docs/architecture/application/workflow.md

@@ -204,14 +204,20 @@ Validate workflows run integrity checks on an existing subscription. Checking th
 ```python
     params = dict(
         name="validate_node_enrollment",
-        target="SYSTEM",
+        target="VALIDATE",
+        is_task=True,
         description="Validate Node Enrollment before production",
         tag="NodeEnrollment",
         search_phrase="Node Enrollment%",
     )
 ```
 
-It uses a `target` of `SYSTEM` - similar to how tasks are defined. That target is more of a free form sort of thing. Same thing with the `name` - that's the name of the actual workflow, and the `tag` is shared by of this set of workflows.
+It uses a `target` of `VALIDATE`. Unlike system tasks, which use the `target` of `SYSTEM` designation, validate
+workflows explicitly use `target="VALIDATE"` to distinguish themselves. This distinction reflects their different
+purposes.
+The `is_task` parameter is set to `True` to indicate that this workflow is a task. Tasks are workflows that are not
+directly associated with a subscription and are typically used for background processing or system maintenance.
+Both `SYSTEM` and `VALIDATE` workflows are considered tasks, but they serve different purposes.
 
 Generally the steps raise assertions if a check fails, otherwise return OK to the state:
 

docs/migration-guide/2.0.md

@@ -1,7 +1,3 @@
----
-hide:
-  - navigation
----
 # 2.0 Migration Guide
 
 In this document we'll help you migrate your orchestrator application from orchestrator-core 1.3 to 2.0.

docs/migration-guide/3.0.md

@@ -1,7 +1,3 @@
----
-hide:
-  - navigation
----
 # 3.0 Migration Guide
 
 In this document we describe the steps that should be taken to migrate from `orchestrator-core` v2 to v3.

docs/migration-guide/4.0.md

@@ -1,16 +1,69 @@
----
-hide:
-  - navigation
----
 # 4.0 Migration Guide
 
 In this document we describe the steps that should be taken to migrate from `orchestrator-core` v3 to v4.
 
 ## About 4.0
 
+### Removed caching of domain models
+
 In this release we have removed the caching of domain models. Domain models will always be loaded from the database.
 
+### Added Target.VALIDATE
+
+In this release, a new workflow target, `VALIDATE`, has been added
+for [validation workflows](../architecture/application/workflow.md#validate).
+Previously, the `SYSTEM` target was used for validation workflows, which implied that they were expected to run in a
+system context. However, this was not appropriate for all validation workflows.
+To address this, the new `VALIDATE` target has been introduced specifically for validation workflows. The `SYSTEM`
+target is now reserved exclusively for system workflows.
+
+The change of the `SYSTEM` target to `VALIDATE` is a breaking change, as it will break any workflows that are using the
+`SYSTEM` target for validation workflows. You will need to update your workflows to use the `VALIDATE` target instead.
+
+In the Steps section below we describe how to update your workflows to use the new `VALIDATE` target.
+
 ## Steps
 
-To use 4.0 all workflows must have run to completion. The `cache_domain_models` step no longer is part of the codebase
-therfore `in flight` workflows will fail.
+To use 4.0.0, all workflows must have run to completion. The `cache_domain_models` step no longer is part of the codebase
+therefore `in flight` workflows will fail.
+
+After running the migration `(2025-05-08_cdf8758831d4_add_is_task_to_workflow.py)`, the workflow table should look like this:
+
+```sql
+| workflow_id                          | name                       | target | is_task | description                                                             | created_at                        | deleted_at |
+|--------------------------------------|----------------------------|--------|---------|-------------------------------------------------------------------------|-----------------------------------|------------|
+| ded79954-f16e-422b-a204-7770a59757e8 | modify_note                | MODIFY | FALSE   | Modify Note                                                             | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| ca6a76ff-dd4e-4f66-9fb0-cee1878f0d20 | task_clean_up_tasks        | SYSTEM | FALSE   | Clean up old tasks                                                      | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| 40058c3d-0c95-47f4-a75f-93719299c5be | task_resume_workflows      | SYSTEM | FALSE   | Resume all workflows that are stuck on tasks with the status 'waiting'  | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| 33b5520e-85d4-4ca1-8713-d26f7de5b7a5 | task_validate_products     | SYSTEM | FALSE   | Validate products                                                       | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| 94d4889e-6bb6-4724-a9d2-f21696fe6f43 | task_validate_product_type | SYSTEM | FALSE   | Validate all subscriptions of Product Type                              | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| 0c4f3b8d-2a1e-4b5f-9a7c-6d8e0f1b2c3d | validate_some_thing        | SYSTEM | FALSE   | Validate The thing                                                      | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+| f4b0a2c1-5d3e-4c8f-9b6d-7a2e5f3b8c4e | validate_another_thing     | SYSTEM | FALSE   | Validate Another thing                                                  | 2025-05-01 09:57:28.033504 +00:00 | <null>     |
+```
+
+The `target` in this table is no longer valid for the `validate_some_thing` and `validate_another_thing`
+workflows. You will need to update the `target` to `VALIDATE` for these workflows. You will also need to update the
+is_task column to `TRUE` for all targets that are `SYSTEM` or `VALIDATE`. This is because the `is_task` column is used to
+determine if a workflow is a task or not. If the `is_task` column is set to `FALSE`, the workflow will not be run as a task.
+Tasks are `SYSTEM` or `VALIDATE` workflows that are run in the context of a system.
+
+Example on how to update the `target` and `is_task` for all workflows that start with `validate_`:
+
+```sql
+UPDATE workflow
+SET target = 'VALIDATE', is_task = TRUE
+WHERE name LIKE 'validate_%';
+```
+
+Example on how to update the `target` and `is_task` for all workflows that are `SYSTEM` or `VALIDATE`:
+
+```sql
+UPDATE workflow
+SET is_task = TRUE
+WHERE target IN ('SYSTEM', 'VALIDATE');
+```
+
+This will update the `target` for all workflows that are `SYSTEM` or `VALIDATE` and set the `is_task` column to `TRUE`.
+
+This is a breaking change, so you will need to test your workflows after making this change to ensure that they are
+working as expected.

orchestrator/__init__.py

@@ -13,7 +13,7 @@
 
 """This is the orchestrator workflow engine."""
 
-__version__ = "4.0.0rc1"
+__version__ = "4.0.0rc2"
 
 from orchestrator.app import OrchestratorCore
 from orchestrator.settings import app_settings

orchestrator/cli/generator/templates/new_product_migration.j2

@@ -63,24 +63,28 @@ new_workflows = [
     {
         "name": "create_{{ product.variable }}",
         "target": Target.CREATE,
+        "is_task": False,
         "description": "Create {{ product.name }}",
         "product_type": "{{ product.type }}",
     },
     {
         "name": "modify_{{ product.variable }}",
         "target": Target.MODIFY,
+        "is_task": False,
         "description": "Modify {{ product.name }}",
         "product_type": "{{ product.type }}",
     },
     {
         "name": "terminate_{{ product.variable }}",
         "target": Target.TERMINATE,
+        "is_task": False,
         "description": "Terminate {{ product.name }}",
         "product_type": "{{ product.type }}",
     },
     {
         "name": "validate_{{ product.variable }}",
-        "target": Target.SYSTEM,
+        "target": Target.VALIDATE,
+        "is_task": True,
         "description": "Validate {{ product.name }}",
         "product_type": "{{ product.type }}",
     },

orchestrator/cli/migrate_tasks.py

@@ -150,22 +150,22 @@ def create_tasks_migration_wizard() -> tuple[list[dict], list[dict]]:
         - list of task items to add in the migration
         - list of task items to delete in the migration
     """
-    database_tasks = {
-        task.name: task for task in list(db.session.scalars(select(WorkflowTable))) if task.target == Target.SYSTEM
-    }
+    database_tasks = {task.name: task for task in list(db.session.scalars(select(WorkflowTable))) if task.is_task}
     registered_wf_instances = {
         task: cast(Workflow, get_workflow(task)) for task in orchestrator.workflows.ALL_WORKFLOWS.keys()
     }
 
+    is_task = [Target.SYSTEM, Target.VALIDATE]
+
     registered_tasks = dict(
         filter(
-            lambda task: task[1].target == Target.SYSTEM and task[0] in database_tasks.keys(),
+            lambda task: task[1].target in is_task and task[0] in database_tasks.keys(),
             registered_wf_instances.items(),
         )
     )
     available_tasks = dict(
         filter(
-            lambda task: task[1].target == Target.SYSTEM and task[0] not in database_tasks.keys(),
+            lambda task: task[1].target in is_task and task[0] not in database_tasks.keys(),
             registered_wf_instances.items(),
         )
     )

orchestrator/cli/migrate_workflows.py

@@ -11,7 +11,6 @@
 from orchestrator.cli.helpers.input_helpers import _enumerate_menu_keys, _prompt_user_menu, get_user_input
 from orchestrator.cli.helpers.print_helpers import COLOR, noqa_print, print_fmt, str_fmt
 from orchestrator.db import ProductTable, WorkflowTable, db
-from orchestrator.targets import Target
 from orchestrator.workflows import LazyWorkflowInstance, get_workflow
 
 # Workflows are registered via migrations with product type. For every product with the given product_type, there will be an entry in products_workflows.
@@ -183,7 +182,7 @@ def create_workflows_migration_wizard() -> tuple[list[dict], list[dict]]:
     """
     database_workflows = list(db.session.scalars(select(WorkflowTable)))
     registered_workflows = orchestrator.workflows.ALL_WORKFLOWS
-    system_workflow_names = {wf.name for wf in database_workflows if wf.target == Target.SYSTEM}
+    system_workflow_names = {wf.name for wf in database_workflows if wf.is_task}
     registered_non_system_workflows = {k: v for k, v in registered_workflows.items() if k not in system_workflow_names}
 
     unregistered_workflows = [wf for wf in database_workflows if wf.name not in registered_workflows.keys()]

orchestrator/db/models.py

@@ -412,6 +412,8 @@ class WorkflowTable(BaseModel):
     )
     processes = relationship("ProcessTable", cascade="all, delete-orphan", back_populates="workflow")
 
+    is_task = mapped_column(Boolean, nullable=False, server_default=text("false"))
+
     @staticmethod
     def select() -> Select:
         return (

orchestrator/migrations/helpers.py

@@ -135,18 +135,21 @@ def create_workflow(conn: sa.engine.Connection, workflow: dict) -> None:
         >>> workflow = {
             "name": "workflow_name",
             "target": "SYSTEM",
+            "is_task": False,
             "description": "workflow description",
             "product_type": "product_type",
         }
         >>> create_workflow(conn, workflow)
     """
+    if not workflow.get("is_task", False):
+        workflow["is_task"] = False
 
     conn.execute(
         sa.text(
             """
                 WITH new_workflow AS (
-                INSERT INTO workflows(name, target, description)
-                    VALUES (:name, :target, :description)
+                INSERT INTO workflows(name, target, is_task, description)
+                    VALUES (:name, :target, :is_task, :description)
                     ON CONFLICT DO NOTHING
                     RETURNING workflow_id)
                 INSERT
@@ -184,8 +187,8 @@ def create_task(conn: sa.engine.Connection, task: dict) -> None:
     conn.execute(
         sa.text(
             """
-                INSERT INTO workflows(name, target, description)
-                    VALUES (:name, 'SYSTEM', :description)
+                INSERT INTO workflows(name, target, is_task, description)
+                    VALUES (:name, 'SYSTEM', TRUE, :description)
                     ON CONFLICT DO NOTHING
                     RETURNING workflow_id
             """
@@ -206,6 +209,7 @@ def create_workflows(conn: sa.engine.Connection, new: dict) -> None:
                 "workflow_name": {
                     "workflow_id": "f2702074-3203-454c-b298-6dfa7675423d",
                     "target": "CREATE",
+                    "is_task": False,
                     "description": "Workflow description",
                     "tag": "ProductBlockName1",
                     "search_phrase": "Search Phrase%",
@@ -214,12 +218,16 @@ def create_workflows(conn: sa.engine.Connection, new: dict) -> None:
     """
     for name, workflow in new.items():
         workflow["name"] = name
+
+        if not workflow.get("is_task", False):
+            workflow["is_task"] = False
+
         conn.execute(
             sa.text(
                 """
                 WITH new_workflow AS (
-                    INSERT INTO workflows(workflow_id, name, target, description)
-                    VALUES (:workflow_id, :name, :target, :description)
+                    INSERT INTO workflows(workflow_id, name, target, is_task, description)
+                    VALUES (:workflow_id, :name, :target, :is_task, :description)
                     RETURNING workflow_id)
                 INSERT
                 INTO products_workflows (product_id, workflow_id)

orchestrator/migrations/versions/schema/2025-02-20_68d14db1b8da_make_workflow_description_mandatory.py

@@ -14,7 +14,7 @@
 
 # revision identifiers, used by Alembic.
 revision = "68d14db1b8da"
-down_revision = "bac6be6f2b4f"
+down_revision = "fc5c993a4b4a"
 branch_labels = None
 depends_on = None
 

orchestrator/migrations/versions/schema/2025-05-08_161918133bec_add_is_task_to_workflow.py

@@ -0,0 +1,28 @@
+"""Add is_task to workflow.
+
+Revision ID: 161918133bec
+Revises: 68d14db1b8da
+Create Date: 2025-05-08 11:25:51.966410
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "161918133bec"
+down_revision = "68d14db1b8da"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("workflows", sa.Column("is_task", sa.Boolean(), server_default=sa.text("false"), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("workflows", "is_task")
+    # ### end Alembic commands ###

orchestrator/schedules/validate_subscriptions.py

@@ -22,7 +22,7 @@
     get_subscriptions_on_product_table_in_sync,
 )
 from orchestrator.services.workflows import (
-    get_system_product_workflows_for_subscription,
+    get_validation_product_workflows_for_subscription,
     start_validation_workflow_for_workflows,
 )
 from orchestrator.settings import app_settings
@@ -42,14 +42,14 @@ def validate_subscriptions() -> None:
         subscriptions = get_subscriptions_on_product_table_in_sync()
 
     for subscription in subscriptions:
-        system_product_workflows = get_system_product_workflows_for_subscription(subscription)
+        validation_product_workflows = get_validation_product_workflows_for_subscription(subscription)
 
-        if not system_product_workflows:
+        if not validation_product_workflows:
             logger.warning(
                 "SubscriptionTable has no validation workflow",
                 subscription=subscription,
                 product=subscription.product.name,
             )
             break
 
-        start_validation_workflow_for_workflows(subscription=subscription, workflows=system_product_workflows)
+        start_validation_workflow_for_workflows(subscription=subscription, workflows=validation_product_workflows)

orchestrator/schemas/workflow.py

@@ -24,6 +24,7 @@
 class WorkflowBaseSchema(OrchestratorBaseModel):
     name: str
     target: Target
+    is_task: bool = False
     description: str | None = None
     created_at: datetime | None = None
 

orchestrator/services/celery.py

@@ -24,7 +24,7 @@
 from orchestrator.db import ProcessTable, db
 from orchestrator.services.input_state import store_input_state
 from orchestrator.services.processes import create_process, delete_process
-from orchestrator.targets import Target
+from orchestrator.services.workflows import get_workflow_by_name
 from orchestrator.workflows import get_workflow
 from pydantic_forms.types import State
 
@@ -51,7 +51,11 @@ def _celery_start_process(
     if not workflow:
         raise_status(HTTPStatus.NOT_FOUND, "Workflow does not exist")
 
-    task_name = NEW_TASK if workflow.target == Target.SYSTEM else NEW_WORKFLOW
+    wf_table = get_workflow_by_name(workflow.name)
+    if not wf_table:
+        raise_status(HTTPStatus.NOT_FOUND, "Workflow in Database does not exist")
+
+    task_name = NEW_TASK if wf_table.is_task else NEW_WORKFLOW
     trigger_task = get_celery_task(task_name)
     pstat = create_process(workflow_key, user_inputs, user)
     try:
@@ -80,7 +84,11 @@ def _celery_resume_process(
     last_process_status = process.last_status
     workflow = pstat.workflow
 
-    task_name = RESUME_TASK if workflow.target == Target.SYSTEM else RESUME_WORKFLOW
+    wf_table = get_workflow_by_name(workflow.name)
+    if not workflow or not wf_table:
+        raise_status(HTTPStatus.NOT_FOUND, "Workflow does not exist")
+
+    task_name = RESUME_TASK if wf_table.is_task else RESUME_WORKFLOW
     trigger_task = get_celery_task(task_name)
 
     user_inputs = user_inputs or [{}]