☊ Step

👣 Step API

Steps are the atomic units of work in a FlowyML pipeline — pure functions decorated with @step.

🎯 Decorator 💾 Caching 🔄 Retries 🖥️ Resources

@step Decorator Parameters

from flowyml import step

@step(
    outputs=["model"],
    inputs=["data"],
    cache=True,
    retries=3,
    timeout=600,
)
def train(data):
    ...

Parameter	Type	Default	Description
outputs	list[str]	[]	Named artifacts this step produces. Used to wire the DAG.
inputs	list[str]	[]	Named artifacts this step consumes. Automatically resolved from upstream steps.
cache	bool	True	Enable content-addressed caching. Skips re-execution when inputs haven't changed.
retries	int	0	Number of automatic retries on failure before the step is marked as failed.
timeout	int | None	None	Maximum execution time in seconds. None means no limit.
resources	dict | None	None	Resource hints (e.g., {"gpu": 1, "memory": "8Gi"}). Interpreted by the orchestrator.
tags	list[str]	[]	Arbitrary tags for filtering and grouping in the UI.
description	str | None	None	Human-readable description shown in the UI DAG view.

Step Function Patterns

Single Output

@step(outputs=["cleaned_data"])
def clean(raw_data):
    """Return value is automatically mapped to 'cleaned_data'."""
    return raw_data.dropna()

Multiple Outputs

@step(outputs=["X_train", "X_test", "y_train", "y_test"])
def split(data):
    """Return a tuple — values are mapped to outputs in order."""
    X_train, X_test, y_train, y_test = train_test_split(data)
    return X_train, X_test, y_train, y_test

Context Injection

@step(outputs=["data"])
def load(dataset_path: str = "default.csv"):
    """Parameters with defaults are auto-injected from context."""
    return pd.read_csv(dataset_path)

How Context Injection Works

If the pipeline's context contains a key matching a step parameter name, the context value is injected automatically. The function default is used as a fallback.

No Outputs (Side-Effect Steps)

@step()
def notify(model_name: str = "model"):
    """Steps with no outputs are valid — useful for notifications or logging."""
    send_slack_message(f"Model {model_name} training complete!")

Common Examples

Cached Training Step with Retries

@step(
    inputs=["X_train", "y_train"],
    outputs=["model"],
    cache=True,
    retries=2,
    timeout=3600,
    resources={"gpu": 1, "memory": "16Gi"},
)
def train_model(X_train, y_train, epochs: int = 50, lr: float = 0.001):
    model = build_model(lr=lr)
    model.fit(X_train, y_train, epochs=epochs)
    return model

Dynamic Step with map_task

from flowyml import step, map_task

@step(outputs=["predictions"])
def predict_batch(model, batch):
    return model.predict(batch)

# Map across multiple batches in parallel
pipeline.add_step(map_task(predict_batch, items=batches))

When to Disable Caching

Disable caching (cache=False) for steps that depend on external state — e.g., fetching live data from an API or reading the latest file from a bucket.

Autodoc

Decorator @step

Decorator to define a pipeline step with automatic context injection.

Can be used as @step or @step(inputs=...)

Every decorated function is automatically registered in a global StepRegistry, enabling Pipeline(auto_discover=True) to build the DAG without any manual add_step() calls.

Parameters:

Name	Type	Description	Default
_func	Callable | None	Function being decorated (when used as @step)	None
inputs	list[str] | None	List of input asset names	None
outputs	list[str] | None	List of output asset names	None
cache	bool | str | Callable	Caching strategy ("code_hash", "input_hash", callable, or False)	'code_hash'
retry	int	Number of retry attempts on failure	0
timeout	int | None	Maximum execution time in seconds	None
resources	Union[dict[str, Any], ResourceRequirements, None]	Resource requirements (ResourceRequirements object or dict for backward compat)	None
tags	dict[str, str] | None	Metadata tags for the step	None
name	str | None	Optional custom name for the step	None
condition	Callable | None	Optional callable that returns True if step should run	None
execution_group	str | None	Optional group name for executing multiple steps together	None
pipeline	str | None	Optional pipeline name for scoped auto-discovery. When set, the step is only auto-discovered by pipelines that match this name (or by get_registered_steps(pipeline="...")).	None
register	bool	If False, the step is NOT added to the global registry. Defaults to True. Set to False for helper/utility steps that should only be used via explicit add_step().	True

Example

@step ... def simple_step(): ... ... @step(inputs=["data/train"], outputs=["model/trained"]) ... def train_model(train_data): ... ...

Scoped to a specific pipeline

@step(pipeline="training", outputs=["model"]) ... def train(data): ... ...

With resource requirements

from flowyml.core.resources import ResourceRequirements, GPUConfig @step(resources=ResourceRequirements(cpu="4", memory="16Gi", gpu=GPUConfig(gpu_type="nvidia-v100", count=2))) ... def gpu_train(data): ... ...

Source code in flowyml/core/step.py

def step(
    _func: Callable | None = None,
    *,
    inputs: list[str] | None = None,
    outputs: list[str] | None = None,
    cache: bool | str | Callable = "code_hash",
    retry: int = 0,
    timeout: int | None = None,
    resources: Union[dict[str, Any], "ResourceRequirements", None] = None,
    tags: dict[str, str] | None = None,
    name: str | None = None,
    condition: Callable | None = None,
    execution_group: str | None = None,
    pipeline: str | None = None,
    register: bool = True,
):
    """Decorator to define a pipeline step with automatic context injection.

    Can be used as @step or @step(inputs=...)

    Every decorated function is automatically registered in a global
    ``StepRegistry``, enabling ``Pipeline(auto_discover=True)`` to build
    the DAG without any manual ``add_step()`` calls.

    Args:
        _func: Function being decorated (when used as @step)
        inputs: List of input asset names
        outputs: List of output asset names
        cache: Caching strategy ("code_hash", "input_hash", callable, or False)
        retry: Number of retry attempts on failure
        timeout: Maximum execution time in seconds
        resources: Resource requirements (ResourceRequirements object or dict for backward compat)
        tags: Metadata tags for the step
        name: Optional custom name for the step
        condition: Optional callable that returns True if step should run
        execution_group: Optional group name for executing multiple steps together
        pipeline: Optional pipeline name for scoped auto-discovery.
            When set, the step is only auto-discovered by pipelines
            that match this name (or by ``get_registered_steps(pipeline="...")``).
        register: If False, the step is NOT added to the global registry.
            Defaults to True. Set to False for helper/utility steps that
            should only be used via explicit ``add_step()``.

    Example:
        >>> @step
        ... def simple_step():
        ...     ...
        >>> @step(inputs=["data/train"], outputs=["model/trained"])
        ... def train_model(train_data):
        ...     ...
        >>> # Scoped to a specific pipeline
        >>> @step(pipeline="training", outputs=["model"])
        ... def train(data):
        ...     ...
        >>> # With resource requirements
        >>> from flowyml.core.resources import ResourceRequirements, GPUConfig
        >>> @step(resources=ResourceRequirements(cpu="4", memory="16Gi", gpu=GPUConfig(gpu_type="nvidia-v100", count=2)))
        ... def gpu_train(data):
        ...     ...
    """

    def decorator(func: Callable) -> Step:
        # Merge pipeline tag into tags dict
        merged_tags = dict(tags) if tags else {}
        if pipeline is not None:
            merged_tags["pipeline"] = pipeline

        step_instance = Step(
            func=func,
            name=name,
            inputs=inputs,
            outputs=outputs,
            cache=cache,
            retry=retry,
            timeout=timeout,
            resources=resources,
            tags=merged_tags if merged_tags else None,
            condition=condition,
            execution_group=execution_group,
        )

        # Auto-register in global registry
        if register:
            _global_registry.register(step_instance)

        return step_instance

    if _func is None:
        return decorator
    else:
        return decorator(_func)

Class Step

A pipeline step that can be executed with automatic context injection.

Source code in flowyml/core/step.py

def __init__(
    self,
    func: Callable,
    name: str | None = None,
    inputs: list[str] | None = None,
    outputs: list[str] | None = None,
    cache: bool | str | Callable = "code_hash",
    retry: int = 0,
    timeout: int | None = None,
    resources: Union[dict[str, Any], "ResourceRequirements", None] = None,
    tags: dict[str, str] | None = None,
    condition: Callable | None = None,
    execution_group: str | None = None,
):
    self.func = func
    self.name = name or func.__name__
    self.inputs = inputs or []
    self.outputs = outputs or []
    self.cache = cache
    self.retry = retry
    self.timeout = timeout

    # Store resources (accept both dict for backward compatibility and ResourceRequirements)
    self.resources = resources
    if self.resources and ResourceRequirements and not isinstance(self.resources, ResourceRequirements):
        if isinstance(self.resources, dict):
            resource_kwargs = dict(self.resources)
            gpu_value = resource_kwargs.get("gpu")
            if GPUConfig and gpu_value is not None:
                if isinstance(gpu_value, dict):
                    resource_kwargs["gpu"] = GPUConfig(
                        gpu_type=gpu_value.get("gpu_type") or gpu_value.get("type") or "generic",
                        count=int(gpu_value.get("count", 1)),
                        memory=gpu_value.get("memory"),
                    )
                elif isinstance(gpu_value, (int, float)):
                    resource_kwargs["gpu"] = GPUConfig(gpu_type="generic", count=int(gpu_value))
            with contextlib.suppress(TypeError):
                self.resources = ResourceRequirements(**resource_kwargs)

    self.tags = tags or {}
    self.condition = condition
    self.execution_group = execution_group

    # Capture source code and location for UI display
    try:
        self.source_code = inspect.getsource(func)
        self.source_file = inspect.getsourcefile(func)
        _, self.source_line = inspect.getsourcelines(func)
    except (OSError, TypeError):
        self.source_code = "# Source code not available"
        self.source_file = None
        self.source_line = None

    self.config = StepConfig(
        name=self.name,
        func=func,
        inputs=self.inputs,
        outputs=self.outputs,
        cache=self.cache,
        retry=self.retry,
        timeout=self.timeout,
        resources=self.resources,
        tags=self.tags,
        condition=self.condition,
        execution_group=self.execution_group,
        source_file=self.source_file,
        source_line=self.source_line,
    )

Functions

__call__(*args: Any, **kwargs: Any) -> Any

Execute the step function.

Source code in flowyml/core/step.py

def __call__(self, *args: Any, **kwargs: Any) -> Any:
    """Execute the step function."""
    # Check condition if present
    if self.condition:
        # We might need to inject context into condition too,
        # but for now assume it takes no args or same args as step?
        # This is tricky without context injection logic here.
        # The executor handles execution, so maybe we just store it here.
        pass

    return self.func(*args, **kwargs)

get_cache_key(inputs: dict[str, Any] | None = None, context_params: dict[str, Any] | None = None) -> str

Generate cache key based on caching strategy.

Parameters:

Name	Type	Description	Default
inputs	dict[str, Any] | None	Input data for the step	None
context_params	dict[str, Any] | None	Context parameters injected into the step. Included in the cache key to prevent stale cache hits when the same step runs with different context values.	None

Returns:

Type	Description
str	Cache key string

Source code in flowyml/core/step.py

def get_cache_key(
    self,
    inputs: dict[str, Any] | None = None,
    context_params: dict[str, Any] | None = None,
) -> str:
    """Generate cache key based on caching strategy.

    Args:
        inputs: Input data for the step
        context_params: Context parameters injected into the step.
            Included in the cache key to prevent stale cache hits
            when the same step runs with different context values.

    Returns:
        Cache key string
    """
    # Compute context suffix so that different context values produce
    # different cache keys for the same step name + code.
    ctx_suffix = ""
    if context_params:
        ctx_str = json.dumps(context_params, sort_keys=True, default=str)
        ctx_suffix = ":" + hashlib.sha256(ctx_str.encode()).hexdigest()[:12]

    if self.cache == "code_hash":
        return f"{self.name}:{self.get_code_hash()}{ctx_suffix}"
    elif self.cache == "input_hash" and inputs:
        return f"{self.name}:{self.get_input_hash(inputs)}{ctx_suffix}"
    elif callable(self.cache) and inputs:
        return self.cache(inputs, {})
    else:
        return f"{self.name}:no-cache"

get_code_hash() -> str

Compute hash of the step's source code.

Source code in flowyml/core/step.py

def get_code_hash(self) -> str:
    """Compute hash of the step's source code."""
    try:
        source = inspect.getsource(self.func)
        return hashlib.md5(source.encode()).hexdigest()
    except (OSError, TypeError):
        # Fallback for dynamically defined functions or when source is unavailable
        return hashlib.md5(self.name.encode()).hexdigest()[:16]

get_input_hash(inputs: dict[str, Any]) -> str

Generate hash of inputs for caching.

Source code in flowyml/core/step.py

def get_input_hash(self, inputs: dict[str, Any]) -> str:
    """Generate hash of inputs for caching."""
    input_str = json.dumps(inputs, sort_keys=True, default=str)
    return hashlib.sha256(input_str.encode()).hexdigest()[:16]

---

🚀 What's Next?

🎢 Pipeline API

Wire steps into a DAG and execute with Pipeline.run().

View Pipeline API →

📜 Context API

Inject parameters into steps with environment-specific configs.

View Context API →

📦 Assets API

Wrap step outputs as Model, Dataset, or Metrics artifacts.

View Assets API →