Add DependentString model and integrate into NodeTemplate and GraphTemplate

- Introduced a new DependentString model to manage dependent values and their relationships.
- Updated NodeTemplate to include a method for generating dependent strings from input values.
- Enhanced GraphTemplate with validation for input dependencies using the new DependentString model.
- Refactored existing code to utilize the new model, improving clarity and maintainability.

These changes enhance the handling of dependent values within templates, ensuring better validation and structure in the graph management process.

Author: NiveditJain

state-manager/app/models/db/graph_template_model.py

@@ -10,6 +10,7 @@
 from ..graph_template_validation_status import GraphTemplateValidationStatus
 from ..node_template_model import NodeTemplate
 from app.utils.encrypter import get_encrypter
+from app.models.dependent_string import DependentString
 
 
 class GraphTemplate(BaseDatabaseModel):
@@ -204,6 +205,31 @@ def verify_unites_identifiers_exist(self) -> Self:
         if errors:
             raise ValueError("\n".join(errors))
         return self
+    
+    @model_validator(mode='after')
+    def verify_input_dependencies(self) -> Self:
+        errors = []
+
+        for node in self.nodes:
+            for input_value in node.inputs.values():
+                try:
+                    dependent_string = DependentString.create_dependent_string(input_value)
+                    dependent_identifiers = set([identifier for identifier, _ in dependent_string.get_identifier_field()])
+
+                    for identifier in dependent_identifiers:
+                        if node.unites is not None:
+                            if identifier not in self.get_parents_by_identifier(node.unites.identifier):
+                                errors.append(f"Input {input_value} depends on {identifier} but {identifier} is not a parent of unites {node.unites.identifier}")
+                        else:
+                            if identifier not in self.get_parents_by_identifier(node.identifier):
+                                errors.append(f"Input {input_value} depends on {identifier} but {identifier} is not a parent of {node.identifier}")
+
+                except Exception as e:
+                    errors.append(f"Error creating dependent string for input {input_value}: {e}")
+        if errors:
+            raise ValueError("\n".join(errors))
+
+        return self
 
     def set_secrets(self, secrets: Dict[str, str]) -> "GraphTemplate":
         self.secrets = {secret_name: get_encrypter().encrypt(secret_value) for secret_name, secret_value in secrets.items()}

state-manager/app/models/dependent_string.py

@@ -0,0 +1,64 @@
+from pydantic import BaseModel, PrivateAttr
+
+class Dependent(BaseModel):
+    identifier: str
+    field: str
+    tail: str
+    value: str | None = None
+
+class DependentString(BaseModel):
+    head: str
+    dependents: dict[int, Dependent]
+    _mapping_key_to_dependent: dict[tuple[str, str], list[Dependent]] = PrivateAttr(default_factory=dict)
+    
+    def generate_string(self) -> str:
+        base = self.head
+        for key in sorted(self.dependents.keys()):
+            dependent = self.dependents[key]
+            if dependent.value is None:
+                raise ValueError(f"Dependent value is not set for: {dependent}")
+            base += dependent.value + dependent.tail
+        return base
+    
+    @staticmethod
+    def create_dependent_string(syntax_string: str) -> "DependentString":
+        splits = syntax_string.split("${{")
+        if len(splits) <= 1:
+            return DependentString(head=syntax_string, dependents={})
+
+        dependent_string = DependentString(head=splits[0], dependents={})
+        order = 0
+
+        for split in splits[1:]:
+            if "}}" not in split:
+                raise ValueError(f"Invalid syntax string placeholder {split} for: {syntax_string} '${{' not closed")
+            placeholder_content, tail = split.split("}}")
+
+            parts = [p.strip() for p in placeholder_content.split(".")]
+            if len(parts) != 3 or parts[1] != "outputs":
+                raise ValueError(f"Invalid syntax string placeholder {placeholder_content} for: {syntax_string}")
+            
+            dependent_string.dependents[order] = Dependent(identifier=parts[0], field=parts[2], tail=tail)
+            order += 1
+
+        return dependent_string
+
+    def _build_mapping_key_to_dependent(self):
+        if self._mapping_key_to_dependent is not None:
+            return
+        
+        for dependent in self.dependents.values():
+            mapping_key = (dependent.identifier, dependent.field)
+            if mapping_key not in self._mapping_key_to_dependent:
+                self._mapping_key_to_dependent[mapping_key] = []
+            self._mapping_key_to_dependent[mapping_key].append(dependent)
+
+    def set_value(self, identifier: str, field: str, value: str):
+        self._build_mapping_key_to_dependent()
+        mapping_key = (identifier, field)
+        for dependent in self._mapping_key_to_dependent[mapping_key]:
+            dependent.value = value
+
+    def get_identifier_field(self) -> list[tuple[str, str]]:
+        self._build_mapping_key_to_dependent()
+        return list(self._mapping_key_to_dependent.keys())

state-manager/app/models/node_template_model.py

@@ -1,5 +1,6 @@
 from pydantic import Field, BaseModel, field_validator
 from typing import Any, Optional, List
+from .dependent_string import DependentString
 
 
 class Unites(BaseModel):
@@ -57,4 +58,10 @@ def validate_unites(cls, v: Optional[Unites]) -> Optional[Unites]:
         if v is not None:
             if v.identifier == "" or v.identifier is None:
                 raise ValueError("Unites identifier cannot be empty")
-        return v
+        return v
+    
+    def get_dependent_strings(self) -> list[DependentString]:
+        dependent_strings = []
+        for input_value in self.inputs.values():
+            dependent_strings.append(DependentString.create_dependent_string(input_value))
+        return dependent_strings

state-manager/app/tasks/create_next_states.py

@@ -7,31 +7,13 @@
 from app.models.state_status_enum import StateStatusEnum
 from app.models.node_template_model import NodeTemplate
 from app.models.db.registered_node import RegisteredNode
+from app.models.dependent_string import DependentString
 from json_schema_to_pydantic import create_model
 from pydantic import BaseModel
 from typing import Type
 
 logger = LogsManager().get_logger()
 
-class Dependent(BaseModel):
-    identifier: str
-    field: str
-    tail: str
-    value: str | None = None
-
-class DependentString(BaseModel):
-    head: str
-    dependents: dict[int, Dependent]
-    
-    def generate_string(self) -> str:
-        base = self.head
-        for key in sorted(self.dependents.keys()):
-            dependent = self.dependents[key]
-            if dependent.value is None:
-                raise ValueError(f"Dependent value is not set for: {dependent}")
-            base += dependent.value + dependent.tail
-        return base
-
 async def mark_success_states(state_ids: list[PydanticObjectId]):
     await State.find(
         In(State.id, state_ids)
@@ -60,36 +42,14 @@ async def check_unites_satisfied(namespace: str, graph_name: str, node_template:
     return True
 
 
-def get_dependents(syntax_string: str) -> DependentString:
-    splits = syntax_string.split("${{")
-    if len(splits) <= 1:
-        return DependentString(head=syntax_string, dependents={})
-
-    dependent_string = DependentString(head=splits[0], dependents={})
-    order = 0
-
-    for split in splits[1:]:
-        if "}}" not in split:
-            raise ValueError(f"Invalid syntax string placeholder {split} for: {syntax_string} '${{' not closed")
-        placeholder_content, tail = split.split("}}")
-
-        parts = [p.strip() for p in placeholder_content.split(".")]
-        if len(parts) != 3 or parts[1] != "outputs":
-            raise ValueError(f"Invalid syntax string placeholder {placeholder_content} for: {syntax_string}")
-        
-        dependent_string.dependents[order] = Dependent(identifier=parts[0], field=parts[2], tail=tail)
-        order += 1
-
-    return dependent_string
-
 def validate_dependencies(next_state_node_template: NodeTemplate, next_state_input_model: Type[BaseModel], identifier: str, parents: dict[str, State]) -> None:
     """Validate that all dependencies exist before processing them."""
     # 1) Confirm each model field is present in next_state_node_template.inputs
     for field_name in next_state_input_model.model_fields.keys():
         if field_name not in next_state_node_template.inputs:
             raise ValueError(f"Field '{field_name}' not found in inputs for template '{next_state_node_template.identifier}'")
 
-        dependency_string = get_dependents(next_state_node_template.inputs[field_name])
+        dependency_string = DependentString.create_dependent_string(next_state_node_template.inputs[field_name])
 
         for dependent in dependency_string.dependents.values():
             # 2) For each placeholder, verify the identifier is either current or present in parents
@@ -110,16 +70,16 @@ def generate_next_state(next_state_input_model: Type[BaseModel], next_state_node
     next_state_input_data = {}
 
     for field_name, _ in next_state_input_model.model_fields.items():
-        dependency_string = get_dependents(next_state_node_template.inputs[field_name])
-
-        for key in sorted(dependency_string.dependents.keys()):
-                if dependency_string.dependents[key].identifier == current_state.identifier:
-                    if dependency_string.dependents[key].field not in current_state.outputs:
-                        raise AttributeError(f"Output field '{dependency_string.dependents[key].field}' not found on current state '{current_state.identifier}' for template '{next_state_node_template.identifier}'")
-                    dependency_string.dependents[key].value = current_state.outputs[dependency_string.dependents[key].field]
-                else:
-                    dependency_string.dependents[key].value = parents[dependency_string.dependents[key].identifier].outputs[dependency_string.dependents[key].field]
-                    
+        dependency_string = DependentString.create_dependent_string(next_state_node_template.inputs[field_name])
+
+        for identifier, field in dependency_string.get_identifier_field():
+            if identifier == current_state.identifier:
+                if field not in current_state.outputs:
+                    raise AttributeError(f"Output field '{field}' not found on current state '{current_state.identifier}' for template '{next_state_node_template.identifier}'")
+                dependency_string.set_value(identifier, field, current_state.outputs[field])
+            else:
+                dependency_string.set_value(identifier, field, parents[identifier].outputs[field])
+                
         next_state_input_data[field_name] = dependency_string.generate_string()
 
     new_parents = {