Fix workflow id and name usage

Related to keephq#1885

Update workflow parsing to use `id` as id and `name` as name, generate different UUID if `id` is not provided, and add validation for configuration errors.

* **keep/parser/parser.py**
  - Update `_get_workflow_id` to use `workflow.get("id")` for id and `workflow.get("name")` for name.
  - Add logic to generate a different UUID if `workflow.get("id")` is not provided.
  - Update `_parse_workflow` to use `workflow.get("name")` for name.

* **keep/api/models/db/workflow.py**
  - Add a new field `invalid` to mark workflows with configuration errors.
  - Update the `Workflow` class to include the `invalid` field.

* **keep/api/routes/workflows.py**
  - Update `create_workflow` and `create_workflow_from_body` to generate a different UUID if `workflow.get("id")` is not provided.
  - Add validation logic to mark workflows with configuration errors as `invalid`.

* **tests/test_parser.py**
  - Add test cases to verify that a different UUID is generated if `workflow.get("id")` is not provided.
  - Add test cases to verify that workflows with configuration errors are marked as `invalid`.

keep/api/models/db/workflow.py

@@ -21,6 +21,7 @@ class Workflow(SQLModel, table=True):
     last_updated: datetime = Field(default_factory=datetime.utcnow)
     provisioned: bool = Field(default=False)
     provisioned_file: Optional[str] = None
+    invalid: bool = Field(default=False)  # P22f5
 
     class Config:
         orm_mode = True

keep/api/routes/workflows.py

@@ -316,6 +316,8 @@ async def create_workflow(
     workflowstore = WorkflowStore()
     # Create the workflow
     try:
+        if not workflow.get("id"):
+            workflow["id"] = str(uuid.uuid4())
         workflow = workflowstore.create_workflow(
             tenant_id=tenant_id, created_by=created_by, workflow=workflow
         )
@@ -355,6 +357,8 @@ async def create_workflow_from_body(
     workflowstore = WorkflowStore()
     # Create the workflow
     try:
+        if not workflow.get("id"):
+            workflow["id"] = str(uuid.uuid4())
         workflow = workflowstore.create_workflow(
             tenant_id=tenant_id, created_by=created_by, workflow=workflow
         )
@@ -379,9 +383,9 @@ async def create_workflow_from_body(
 
 # Add Mock Workflows (6 Random Workflows on Every Request)
 #    To add mock workflows, a new backend API endpoint has been created: /workflows/random-templates.
-#      1. Fetching Random Templates: When a request is made to this endpoint, all workflow YAML/YML files are read and
-#         shuffled randomly.
-#      2. Response: Only the first 6 files are parsed and sent in the response.
+#       1. Fetching Random Templates: When a request is made to this endpoint, all workflow YAML/YML files are read and
+#          shuffled randomly.
+#       2. Response: Only the first 6 files are parsed and sent in the response.
 @router.get("/random-templates", description="Get random workflow templates")
 def get_random_workflow_templates(
     authenticated_entity: AuthenticatedEntity = Depends(

keep/parser/parser.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import typing
+import uuid
 
 import yaml
 
@@ -32,20 +33,16 @@ def _get_workflow_id(self, tenant_id, workflow: dict) -> str:
         Returns:
             str: _description_
         """
-        # for backward compatibility reasons, the id on the YAML is actually the name
-        # and the id is a unique generated id stored in the db
-        workflow_name = workflow.get("id")
+        workflow_id = workflow.get("id")
+        workflow_name = workflow.get("name")
+        if workflow_id is None:
+            workflow_id = str(uuid.uuid4())
         if workflow_name is None:
-            raise ValueError("Workflow dict must have an id")
-
-        # get the workflow id from the database
-        workflow_id = get_workflow_id(tenant_id, workflow_name)
-        # if the workflow id is not found, it means that the workflow is not stored in the db
-        # for example when running from CLI
-        # so for backward compatibility, we will use the workflow name as the id
-        # todo - refactor CLI to use db also
-        if not workflow_id:
-            workflow_id = workflow_name
+            raise ValueError("Workflow dict must have a name")
+
+        db_workflow_id = get_workflow_id(tenant_id, workflow_name)
+        if db_workflow_id:
+            workflow_id = db_workflow_id
         return workflow_id
 
     def parse(

tests/test_parser.py

@@ -414,3 +414,37 @@ def test_deep_merge_list(self):
 
         results = ParserUtils.deep_merge(source, dest)
         assert expected_results == results
+
+
+class TestWorkflowUUIDGeneration:
+    def test_generate_different_uuid_if_id_not_provided(self):
+        parser = Parser()
+        workflow = {"name": "test_workflow"}
+        tenant_id = "test_tenant"
+        generated_id = parser._get_workflow_id(tenant_id, workflow)
+        assert generated_id is not None
+        assert isinstance(generated_id, str)
+        assert len(generated_id) == 36  # UUID length
+
+    def test_use_provided_id(self):
+        parser = Parser()
+        workflow = {"id": "test_id", "name": "test_workflow"}
+        tenant_id = "test_tenant"
+        generated_id = parser._get_workflow_id(tenant_id, workflow)
+        assert generated_id == "test_id"
+
+
+class TestWorkflowInvalidFlag:
+    def test_workflow_invalid_flag(self):
+        parser = Parser()
+        workflow = {"name": "test_workflow", "invalid": True}
+        tenant_id = "test_tenant"
+        parsed_workflow = parser._parse_workflow(tenant_id, workflow, None)
+        assert parsed_workflow.invalid is True
+
+    def test_workflow_valid_flag(self):
+        parser = Parser()
+        workflow = {"name": "test_workflow", "invalid": False}
+        tenant_id = "test_tenant"
+        parsed_workflow = parser._parse_workflow(tenant_id, workflow, None)
+        assert parsed_workflow.invalid is False