Skip to content

📜 Context

📜 Context API

Context separates configuration from code — inject parameters into steps without hard-coding values.

🎛️ Parameters 🌍 Environments 🔀 Overrides

context() Function

from flowyml import context

ctx = context(
    dataset_path="data/train.csv",
    epochs=50,
    learning_rate=0.001,
)
Parameter Type Default Description
**kwargs Any — Arbitrary key-value pairs. Keys matching step parameter names are injected automatically.

Immutable by Design

Once created, a context object is immutable. Use ctx.override() to create a new context with updated values.

Context Methods & Properties

Method / Property Signature Description
override override(**kwargs) → Context Return a new context with specified keys replaced. Original is unchanged.
merge merge(other: Context) → Context Combine two contexts. Values from other take precedence on conflict.
to_dict to_dict() → dict Serialize all parameters to a plain dictionary.
get get(key, default=None) Retrieve a single parameter value with an optional default.
pipeline_name str (property) Name of the pipeline this context is bound to (set at runtime).
run_id str (property) Unique identifier for the current pipeline run.
environment str (property) Current environment name (e.g., dev, staging, prod).

Usage Examples

Basic Context

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
from flowyml import Pipeline, step, context

@step(outputs=["data"])
def load(dataset_path: str = "default.csv"):
    return pd.read_csv(dataset_path)

ctx = context(dataset_path="gs://bucket/train.csv")
pipeline = Pipeline("basic", context=ctx)
pipeline.add_step(load)
pipeline.run()  # load receives dataset_path="gs://bucket/train.csv"

Environment-Specific Configs

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
import os
from flowyml import context

base = context(
    batch_size=32,
    learning_rate=0.001,
)

if os.getenv("FLOWYML_ENV") == "prod":
    ctx = base.override(
        dataset_path="gs://prod-bucket/data.csv",
        batch_size=256,
    )
else:
    ctx = base.override(
        dataset_path="local/dev_sample.csv",
        batch_size=16,
    )

Runtime Override via CLI

flowyml run pipeline.py \
  --context dataset_path=gs://bucket/new_data.csv \
  --context epochs=100

Override Precedence

Values are resolved in this order (highest priority first):

  1. CLI --context flags — always win
  2. pipeline.run(**overrides) — programmatic overrides
  3. context() constructor — base configuration
  4. Step parameter defaults — fallback values

Accessing Context Inside Steps

1
2
3
4
5
6
7
8
9
from flowyml import step, get_context

@step(outputs=["report"])
def generate_report():
    ctx = get_context()
    print(f"Pipeline: {ctx.pipeline_name}")
    print(f"Run ID:   {ctx.run_id}")
    print(f"Env:      {ctx.environment}")
    return {"run_id": ctx.run_id, "status": "complete"}

Autodoc

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 
14
15
16
17
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 
19
20
21
22
23
24
25
26
27
28
29
30
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 
32
33
34
35
36
37
38
39
40
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 
42
43
44
45
46
47
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 
80
81
82
83
84
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 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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 
56
57
58
59
60
61
62
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 
49
50
51
52
53
54
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 
64
65
66
67
68
69
70
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 
72
73
74
75
76
77
78
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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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

🚀 What's Next?

🎢 Pipeline API

Pass contexts to pipelines and control execution behavior.

View Pipeline API →

👣 Step API

See how step parameters receive injected context values.

View Step API →

📦 Assets API

Artifacts produced by steps — Model, Dataset, and Metrics.

View Assets API →