Context API 🧠

Access runtime information, parameters, and configuration within a step.

Usage

from flowyml import step, get_context

@step
def my_step():
    ctx = get_context()
    print(f"Running in pipeline: {ctx.pipeline_name}")

Class Context

Pipeline context with automatic parameter injection.

Example

ctx = Context(learning_rate=0.001, epochs=10, batch_size=32, device="cuda")

Source code in flowyml/core/context.py

def __init__(self, **kwargs):
    self._params = kwargs
    self._parent = None
    self._metadata = {}

Functions

__getattr__(name: str) -> Any

Allow dot notation access to parameters.

Source code in flowyml/core/context.py

def __getattr__(self, name: str) -> Any:
    """Allow dot notation access to parameters."""
    if name.startswith("_"):
        return super().__getattribute__(name)

    if name in self._params:
        return self._params[name]

    if self._parent and name in self._parent._params:
        return self._parent._params[name]

    raise AttributeError(f"Context has no parameter '{name}'")

__getitem__(key: str) -> Any

Allow dict-style access to parameters.

Source code in flowyml/core/context.py

def __getitem__(self, key: str) -> Any:
    """Allow dict-style access to parameters."""
    if key in self._params:
        return self._params[key]

    if self._parent and key in self._parent._params:
        return self._parent._params[key]

    raise KeyError(f"Context has no parameter '{key}'")

get(key: str, default: Any = None) -> Any

Get parameter with default value.

Source code in flowyml/core/context.py

def get(self, key: str, default: Any = None) -> Any:
    """Get parameter with default value."""
    try:
        return self[key]
    except KeyError:
        return default

inherit(**overrides) -> Context

Create child context with inheritance.

Source code in flowyml/core/context.py

def inherit(self, **overrides) -> "Context":
    """Create child context with inheritance."""
    child = Context(**overrides)
    child._parent = self
    return child

inject_params(func: callable) -> dict[str, Any]

Automatically inject parameters based on function signature.

Parameters:

Name	Type	Description	Default
func	callable	Function to analyze and inject parameters for	required

Returns:

Type	Description
dict[str, Any]	Dictionary of parameters to inject

Source code in flowyml/core/context.py

def inject_params(self, func: callable) -> dict[str, Any]:
    """Automatically inject parameters based on function signature.

    Args:
        func: Function to analyze and inject parameters for

    Returns:
        Dictionary of parameters to inject
    """
    sig = inspect.signature(func)
    injected = {}

    for param_name, param in sig.parameters.items():
        # Skip self, cls, args, kwargs
        if param_name in ("self", "cls"):
            continue
        if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
            continue

        # Check if parameter exists in context
        if param_name in self.keys():
            injected[param_name] = self[param_name]

    return injected

items() -> list[tuple]

Return all parameter items.

Source code in flowyml/core/context.py

def items(self) -> list[tuple]:
    """Return all parameter items."""
    result = {}
    if self._parent:
        result.update(dict(self._parent.items()))
    result.update(self._params)
    return list(result.items())

keys() -> set[str]

Return all parameter keys.

Source code in flowyml/core/context.py

def keys(self) -> set[str]:
    """Return all parameter keys."""
    keys = set(self._params.keys())
    if self._parent:
        keys.update(self._parent.keys())
    return keys

to_dict() -> dict[str, Any]

Convert context to dictionary.

Source code in flowyml/core/context.py

def to_dict(self) -> dict[str, Any]:
    """Convert context to dictionary."""
    result = {}
    if self._parent:
        result.update(self._parent.to_dict())
    result.update(self._params)
    return result

update(data: dict[str, Any]) -> None

Update context with new data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary of key-value pairs to add to context	required

Source code in flowyml/core/context.py

def update(self, data: dict[str, Any]) -> None:
    """Update context with new data.

    Args:
        data: Dictionary of key-value pairs to add to context
    """
    self._params.update(data)

validate_for_step(step_func: callable, exclude: list[str] = None) -> list[str]

Validate that all required parameters are available.

Parameters:

Name	Type	Description	Default
step_func	callable	Step function to validate	required
exclude	list[str]	List of parameter names to exclude from validation (e.g. inputs)	None

Returns:

Type	Description
list[str]	List of missing required parameters

Source code in flowyml/core/context.py

def validate_for_step(self, step_func: callable, exclude: list[str] = None) -> list[str]:
    """Validate that all required parameters are available.

    Args:
        step_func: Step function to validate
        exclude: List of parameter names to exclude from validation (e.g. inputs)

    Returns:
        List of missing required parameters
    """
    sig = inspect.signature(step_func)
    missing = []
    exclude = exclude or []

    for param_name, param in sig.parameters.items():
        # Skip optional parameters
        if param_name in ("self", "cls"):
            continue
        if param_name in exclude:
            continue
        if param.default != inspect.Parameter.empty:
            continue
        if param.kind in (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD):
            continue

        # Check if required param is missing
        if param_name not in self.keys():
            missing.append(param_name)

    return missing