entities.py 1.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. from typing import Any, Optional
  2. from core.workflow.entities.base_node_data_entities import BaseIterationNodeData, BaseIterationState, BaseNodeData
  3. class IterationNodeData(BaseIterationNodeData):
  4. """
  5. Iteration Node Data.
  6. """
  7. parent_loop_id: Optional[str] = None # redundant field, not used currently
  8. iterator_selector: list[str] # variable selector
  9. output_selector: list[str] # output selector
  10. class IterationStartNodeData(BaseNodeData):
  11. """
  12. Iteration Start Node Data.
  13. """
  14. pass
  15. class IterationState(BaseIterationState):
  16. """
  17. Iteration State.
  18. """
  19. outputs: list[Any] = None
  20. current_output: Optional[Any] = None
  21. class MetaData(BaseIterationState.MetaData):
  22. """
  23. Data.
  24. """
  25. iterator_length: int
  26. def get_last_output(self) -> Optional[Any]:
  27. """
  28. Get last output.
  29. """
  30. if self.outputs:
  31. return self.outputs[-1]
  32. return None
  33. def get_current_output(self) -> Optional[Any]:
  34. """
  35. Get current output.
  36. """
  37. return self.current_output