Class Workflow

preswald/interfaces/workflow.py:195–601

Core workflow engine that manages registration and execution of reactive atoms.

                logger.debug(f"[CONTEXT] Set result for {atom_name} = {result.value}")

class Workflow:
    """
    Core workflow engine that manages registration and execution of reactive atoms.
    """

    def __init__(self, service: Optional["BasePreswaldService"] = None, default_retry_policy: Optional[RetryPolicy] = None):
        self.atoms: dict[str, Atom] = {}
        self.context = WorkflowContext()
        self.default_retry_policy = default_retry_policy or RetryPolicy()
        self.cache = AtomCache()
        self._component_producers: dict[str, str] = {}  # component_id -> atom_name
        self._current_atom: str | None = None  # currently executing atom
        self._service = service
        self._is_rerun = False
        self._auto_atom_registry: dict[str, Callable] = {}
        self._registered_reactive_atoms: list[Callable] = []

    def atom(
        self,
        dependencies: list[str] | None = None,
        retry_policy: RetryPolicy | None = None,
        force_recompute: bool = False,
        name: str | None = None,
    ):
        """
        Decorator to manually register a function as a reactive atom in the workflow.

        Atoms can be registered either explicitly using this decorator
        or automatically through code transformation at runtime.

        If dependencies are not explicitly provided, they will be inferred
        from the function's parameter names.

        TODO: provide example usage before PR comes out of draft

        Args:
            dependencies (list[str], optional):
                Explicit list of atom names this atom depends on. If omitted, inferred from function arguments.
            retry_policy (RetryPolicy, optional):
                Custom retry policy to apply when this atom fails.
            force_recompute (bool, optional):
                If True, forces this atom to recompute even if inputs have not changed.
            name (str, optional):
                Custom name for the atom. Defaults to the function's name.
        """
        def decorator(func):
            atom_name = name or func.__name__

            if self._is_rerun and atom_name in self.atoms:
                logger.debug(f"[workflow.atom] Skipping re-registration during rerun {atom_name=}")
                return func

            logger.info(f"[workflow.atom] Registered atom {atom_name=}")

            # Use the unwrapped function to infer dependencies
            raw_func = getattr(func, 'original_func', func)
            inferred_deps = [
                k for k, v in inspect.signature(raw_func).parameters.items()