Skip to content

PipelineMeta

Pipeline Metadata

PipelineMeta resolves pipeline configuration from the PIPELINE_META environment variable.

PipelineMeta reads a JSON pipeline configuration from the PIPELINE_META environment variable and exposes it as attributes (model_name, endpoint_name, mode, serverless, etc.). It fails hard — raising RuntimeError if PIPELINE_META is unset or invalid — so pipeline steps don't run with silently missing config.

PipelineMeta: Resolves pipeline metadata from environment configuration.

PipelineMeta

PipelineMeta: Resolves pipeline metadata from the PIPELINE_META environment variable.

Reads pipeline configuration from the PIPELINE_META environment variable (JSON dict). Raises RuntimeError if PIPELINE_META is not set or contains invalid JSON.

Common Usage 
from workbench.core.pipelines.pipeline_meta import PipelineMeta

pm = PipelineMeta()
model_name = pm.model_name
endpoint_name = pm.endpoint_name
mode = pm.mode
serverless = pm.serverless

# Access arbitrary keys (fails hard if key missing and no default)
custom_value = pm.get("custom_key", default="fallback")
Environment Variable

PIPELINE_META: JSON dict with pipeline configuration, e.g.: 

PIPELINE_META='{"mode": "dt", "model_name": "my-model-dt",
               "endpoint_name": "my-endpoint-dt", "serverless": true}'

Source code in src/workbench/core/pipelines/pipeline_meta.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
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
141
142
143
144
145
146
147
148
149
150
151
class PipelineMeta:
    """PipelineMeta: Resolves pipeline metadata from the PIPELINE_META environment variable.

    Reads pipeline configuration from the PIPELINE_META environment variable (JSON dict).
    Raises RuntimeError if PIPELINE_META is not set or contains invalid JSON.

    Common Usage:
        ```python
        from workbench.core.pipelines.pipeline_meta import PipelineMeta

        pm = PipelineMeta()
        model_name = pm.model_name
        endpoint_name = pm.endpoint_name
        mode = pm.mode
        serverless = pm.serverless

        # Access arbitrary keys (fails hard if key missing and no default)
        custom_value = pm.get("custom_key", default="fallback")
        ```

    Environment Variable:
        PIPELINE_META: JSON dict with pipeline configuration, e.g.:
        ```
        PIPELINE_META='{"mode": "dt", "model_name": "my-model-dt",
                       "endpoint_name": "my-endpoint-dt", "serverless": true}'
        ```
    """

    def __init__(self):
        """Initialize PipelineMeta from the PIPELINE_META environment variable."""
        self.log = logging.getLogger("workbench")
        self._meta = {}
        self._owner = "test"
        self._resolve()

    def get(self, key: str, default: Any = _MISSING) -> Any:
        """Get a value from the pipeline metadata.

        Args:
            key (str): The key to look up
            default: Default value if key is not found (raises RuntimeError if omitted)

        Returns:
            The value for the key, or default if not found
        """
        if key in self._meta:
            return self._meta[key]
        if default is not _MISSING:
            return default
        msg = f"PipelineMeta: Key '{key}' not found in PIPELINE_META"
        self.log.critical(msg)
        raise RuntimeError(msg)

    @property
    def model_name(self) -> str:
        """The resolved model name."""
        return self._meta["model_name"]

    @property
    def endpoint_name(self) -> str:
        """The resolved endpoint name."""
        return self._meta["endpoint_name"]

    @property
    def challengers(self) -> list:
        """The challenger model names for a promote node (its model inputs)."""
        return self._meta["challengers"]

    @property
    def mode(self) -> str | None:
        """The pipeline execution mode (e.g., 'dt', 'ts'), or None for a modeless run."""
        return self._meta.get("mode")

    @property
    def serverless(self) -> bool:
        """Whether to use serverless inference."""
        return self._meta["serverless"]

    def set_owner(self, owner: str):
        """Set the owner for dynamic owner resolution.

        Args:
            owner (str): The owner identifier (e.g., "BW", "Bob")
        """
        self._owner = owner

    def dynamic_owner(self) -> str:
        """Return mode-appropriate owner string.

        Uses the owner set via set_owner() and transforms based on mode:
            - dt / ts: "DT"
            - promote: "Pro-{owner}"
            - any other: "{owner}"

        Returns:
            The resolved owner string
        """
        mode = self.mode
        owner = self._owner
        if mode in ("dt", "ts"):
            return "DT"
        elif mode == "promote":
            return f"Pro-{owner}"
        else:
            return owner

    def _resolve(self):
        """Resolve pipeline metadata from the PIPELINE_META environment variable."""
        pipeline_meta_json = os.environ.get("PIPELINE_META")
        if not pipeline_meta_json:
            msg = "PipelineMeta: PIPELINE_META environment variable not set"
            self.log.critical(msg)
            raise RuntimeError(msg)
        self._resolve_from_env(pipeline_meta_json)

    def _resolve_from_env(self, pipeline_meta_json: str):
        """Parse pipeline metadata from the PIPELINE_META environment variable.

        Args:
            pipeline_meta_json (str): JSON string from PIPELINE_META env var
        """
        try:
            self._meta = json.loads(pipeline_meta_json)
        except json.JSONDecodeError as e:
            msg = f"PipelineMeta: Failed to parse PIPELINE_META: {e}"
            self.log.critical(msg)
            raise RuntimeError(msg)

        # mode defaults to None (modeless nodes have none -- don't fabricate a 'dt'); a
        # known key so pm.mode and pm.get("mode") agree. serverless defaults to True.
        self._meta.setdefault("mode", None)
        self._meta.setdefault("serverless", True)
        self.log.info(f"PipelineMeta: mode={self._meta.get('mode')}, model={self._meta.get('model_name', 'N/A')}")

    def __repr__(self) -> str:
        """String representation of this PipelineMeta."""
        return (
            f"PipelineMeta(mode={self._meta.get('mode')}, model={self._meta.get('model_name')}, "
            f"endpoint={self._meta.get('endpoint_name')}, serverless={self._meta.get('serverless')})"
        )

challengers property

The challenger model names for a promote node (its model inputs).

endpoint_name property

The resolved endpoint name.

mode property

The pipeline execution mode (e.g., 'dt', 'ts'), or None for a modeless run.

model_name property

The resolved model name.

serverless property

Whether to use serverless inference.

__init__()

Initialize PipelineMeta from the PIPELINE_META environment variable.

Source code in src/workbench/core/pipelines/pipeline_meta.py 
40
41
42
43
44
45
def __init__(self):
    """Initialize PipelineMeta from the PIPELINE_META environment variable."""
    self.log = logging.getLogger("workbench")
    self._meta = {}
    self._owner = "test"
    self._resolve()

__repr__()

String representation of this PipelineMeta.

Source code in src/workbench/core/pipelines/pipeline_meta.py 
146
147
148
149
150
151
def __repr__(self) -> str:
    """String representation of this PipelineMeta."""
    return (
        f"PipelineMeta(mode={self._meta.get('mode')}, model={self._meta.get('model_name')}, "
        f"endpoint={self._meta.get('endpoint_name')}, serverless={self._meta.get('serverless')})"
    )

dynamic_owner()

Return mode-appropriate owner string.

Uses the owner set via set_owner() and transforms based on mode: - dt / ts: "DT" - promote: "Pro-{owner}" - any other: "{owner}"

Returns:

Type Description
str

The resolved owner string
Source code in src/workbench/core/pipelines/pipeline_meta.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def dynamic_owner(self) -> str:
    """Return mode-appropriate owner string.

    Uses the owner set via set_owner() and transforms based on mode:
        - dt / ts: "DT"
        - promote: "Pro-{owner}"
        - any other: "{owner}"

    Returns:
        The resolved owner string
    """
    mode = self.mode
    owner = self._owner
    if mode in ("dt", "ts"):
        return "DT"
    elif mode == "promote":
        return f"Pro-{owner}"
    else:
        return owner

get(key, default=_MISSING)

Get a value from the pipeline metadata.

Parameters:

Name Type Description Default
key str

The key to look up

 required
default Any

Default value if key is not found (raises RuntimeError if omitted)

 _MISSING

Returns:

Type Description
Any

The value for the key, or default if not found
Source code in src/workbench/core/pipelines/pipeline_meta.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def get(self, key: str, default: Any = _MISSING) -> Any:
    """Get a value from the pipeline metadata.

    Args:
        key (str): The key to look up
        default: Default value if key is not found (raises RuntimeError if omitted)

    Returns:
        The value for the key, or default if not found
    """
    if key in self._meta:
        return self._meta[key]
    if default is not _MISSING:
        return default
    msg = f"PipelineMeta: Key '{key}' not found in PIPELINE_META"
    self.log.critical(msg)
    raise RuntimeError(msg)

set_owner(owner)

Set the owner for dynamic owner resolution.

Parameters:

Name Type Description Default
owner str

The owner identifier (e.g., "BW", "Bob")

 required
Source code in src/workbench/core/pipelines/pipeline_meta.py 
90
91
92
93
94
95
96
def set_owner(self, owner: str):
    """Set the owner for dynamic owner resolution.

    Args:
        owner (str): The owner identifier (e.g., "BW", "Bob")
    """
    self._owner = owner

Questions?

The SuperCowPowers team is happy to answer any questions you may have about AWS and Workbench. Please contact us at workbench@supercowpowers.com or on chat us up on Discord