diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py index a4fbdca..18460b0 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py @@ -20,6 +20,7 @@ """ from __future__ import annotations +from .dynamic_data import dynamic_secret from .pipeline import pipeline from .raw import raw from .ref import ref @@ -35,6 +36,7 @@ "registered", "ref", "raw", + "dynamic_secret", "subpipeline", "TaskEnv", "In", diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/dynamic_data.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/dynamic_data.py new file mode 100644 index 0000000..5eb1b76 --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/dynamic_data.py @@ -0,0 +1,44 @@ +"""Dynamic-data argument helpers for Python-authored pipelines. + +Tangle runnable arguments can be literals, graph/task edges, or dynamic data +resolved by the runtime (for example a secret reference). The Python pipeline +emitter needs an explicit wrapper so author code can request a dynamicData +argument without opening support for arbitrary dict constants. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + + +@dataclass(frozen=True) +class DynamicData: + """A task argument value emitted as ``{"dynamicData": value}``. + + Internal emit primitive — not part of the public authoring surface (see + ``__init__.__all__``). ``value`` is a general mapping because the runtime + resolves several dynamic-data kinds (secrets, run IDs, loop indices), but + the only public producer today is :func:`dynamic_secret`, which always + builds the ``{"secret": {"name": ...}}`` shape that the strict dehydrated + schema accepts. Any other shape emits but fails closed at compile + validation until a new kind is added to both the constructor surface and + the schema together — so the public API never promises more than the + schema enforces. + """ + + value: Mapping[str, Any] + + +def dynamic_secret(name: str) -> DynamicData: + """Reference a runtime secret by name for a task argument. + + Example emitted YAML:: + + openai_api_key: + dynamicData: + secret: + name: OPENAI_API_KEY + """ + if not isinstance(name, str) or not name: + raise ValueError("dynamic_secret() requires a non-empty secret name string") + return DynamicData({"secret": {"name": name}}) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py index 4bbd3b6..8887ffe 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/emit.py @@ -20,11 +20,14 @@ also recorded in the returned exempt-paths set so the no-template-delimiter output guard skips that one location (it is a legitimate RUNTIME placeholder, e.g. a run-query ``{{input_1}}`` sentinel). +* a :class:`DynamicData` value → ``{"dynamicData": value}`` for runtime- + resolved arguments such as secrets. Non-string constants are rejected: the runnable Tangle argument contract -only supports string constants, ``graphInput``, or ``taskOutput``. -Structured/non-string values must be stringified explicitly in pipeline -code (e.g. ``json.dumps(...)``) before they reach the compiler. +only supports string constants, ``graphInput``, ``taskOutput``, or explicit +``dynamicData`` wrappers. Structured/non-string values must be stringified +explicitly in pipeline code (e.g. ``json.dumps(...)``) before they reach the +compiler. The literal key the user wrote (``wait_for``, ``depends_on``, ``project``, ``payload``, ...) is preserved verbatim as the dict key. @@ -33,6 +36,7 @@ from typing import Any +from .dynamic_data import DynamicData from .errors import CompileError, InvalidArgumentTypeError from .graph import EdgeRef, GraphBuilder, TaskNode from .placeholders import GraphInputPlaceholder, TaskOutputProxy @@ -219,7 +223,8 @@ def _emit_argument_value( Dispatch is purely on the VALUE's runtime type — the argument *key* is never inspected. Produces a ``taskOutput`` / ``graphInput`` wrapper for edges, or the RAW string for a constant (matching the runnable - Tangle argument contract). Non-string constants are rejected. + Tangle argument contract). Non-string constants are rejected unless they + are explicit dynamic-data wrappers. A :class:`Raw` value is emitted as its inner string verbatim — exactly like a plain ``str`` constant — and ``arg_path`` (this argument's @@ -244,6 +249,8 @@ def _emit_argument_value( # recorded here, at the only point the Raw wrapper is still visible. exempt_paths.add(arg_path) return value.value + if isinstance(value, DynamicData): + return {"dynamicData": value.value} # Everything else is a constant. The runnable schema only accepts raw # string constants, so validate and emit the string verbatim. _validate_constant(value, key) @@ -254,10 +261,10 @@ def _validate_constant(value: Any, key: str) -> None: """Assert ``value`` is a runnable string constant. Runnable Tangle pipeline arguments only support raw ``str`` constants - (alongside the ``graphInput`` / ``taskOutput`` wrappers). A non-string - constant (``int``, ``float``, ``bool``, ``None``, ``list``, ``dict``, - a tuple/set, a leftover helper object, a callable, ...) cannot be - represented under the runnable schema, so it is rejected with a + (alongside the ``graphInput`` / ``taskOutput`` / explicit ``dynamicData`` + wrappers). A non-string constant (``int``, ``float``, ``bool``, ``None``, + ``list``, ``dict``, a tuple/set, a leftover helper object, a callable, ...) + cannot be represented under the runnable schema, so it is rejected with a GENERIC, operation-agnostic message — the compiler has no operation-specific knowledge (no SQL, BigQuery, or other domain awareness). Authors must stringify structured/non-string values @@ -268,7 +275,8 @@ def _validate_constant(value: Any, key: str) -> None: raise InvalidArgumentTypeError( f"unsupported constant type {type(value).__name__!r} for " f"argument {key!r}. Runnable Tangle pipeline arguments only support " - "string constants, graphInput, or taskOutput. Convert structured or " + "string constants, graphInput, or taskOutput; dynamicData wrappers are also supported. " + "Convert structured or " "non-string values to a string explicitly in your pipeline code " "(for example json.dumps(...) or str(...)) before passing them as " "task arguments." diff --git a/packages/tangle-cli/src/tangle_cli/schema_validation.py b/packages/tangle-cli/src/tangle_cli/schema_validation.py index 774d058..c33c15c 100644 --- a/packages/tangle-cli/src/tangle_cli/schema_validation.py +++ b/packages/tangle-cli/src/tangle_cli/schema_validation.py @@ -19,7 +19,7 @@ * :func:`is_dehydrated_pipeline` — shape detector (no raise): top-level ``name`` + ``implementation.graph.tasks``, no ``template_file``, and task ``arguments`` values that are raw string constants or ``graphInput`` - / ``taskOutput`` wrappers. + / ``taskOutput`` / ``dynamicData`` wrappers. * :func:`validate_dehydrated_pipeline` — JSON-Schema validation PLUS the deeper semantic checks jsonschema cannot express cleanly (dangling ``taskOutput.taskId``, undeclared ``graphInput.inputName``, @@ -207,25 +207,25 @@ def assert_no_template_delimiters( {"name", "description", "metadata", "inputs", "outputs", "implementation"} ) -# The reference-only ArgumentValue wrappers. A constant is NOT a wrapper — -# it is a raw string (matching the runnable Tangle argument contract). -_REFERENCE_ARGUMENT_KEYS = ("graphInput", "taskOutput") +# ArgumentValue wrappers. A constant is NOT a wrapper — it is a raw string +# (matching the runnable Tangle argument contract). +_ARGUMENT_WRAPPER_KEYS = ("graphInput", "taskOutput", "dynamicData") def _is_argument_value(value: Any) -> bool: """True when ``value`` looks like a runnable ArgumentValue — a raw - string constant, or a mapping carrying a ``graphInput`` / ``taskOutput`` - wrapper. + string constant, or a mapping carrying a ``graphInput`` / ``taskOutput`` / + ``dynamicData`` wrapper. There is no ambiguity: a raw string constant (even one whose text is ``"graphInput"`` or JSON like ``'{"graphInput": ...}'``) is a string, never the object wrapper shapes — so it can never collide with a - ``graphInput`` / ``taskOutput`` mapping. + ``graphInput`` / ``taskOutput`` / ``dynamicData`` mapping. """ if isinstance(value, str): return True return isinstance(value, Mapping) and any( - key in value for key in _REFERENCE_ARGUMENT_KEYS + key in value for key in _ARGUMENT_WRAPPER_KEYS ) @@ -238,7 +238,7 @@ def is_dehydrated_pipeline(data: Any) -> bool: * NO ``template_file`` (it is the final rendered form, not a wrapper); * task ``arguments`` values AND graph ``outputValues`` values (when present) that are raw string constants or ``graphInput`` / - ``taskOutput`` wrappers — a non-string raw value (a bare + ``taskOutput`` / ``dynamicData`` wrappers — a non-string raw value (a bare number/list/object) or a legacy ``{constantValue: ...}`` wrapper is not a runnable argument value, so it means the input is not yet dehydrated. diff --git a/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json b/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json index 47fcc8a..fdf311c 100644 --- a/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json +++ b/packages/tangle-cli/src/tangle_cli/schemas/dehydrated_pipeline_schema.json @@ -169,11 +169,12 @@ }, "ArgumentValue": { - "description": "Source of a task argument or graph output. Each argument has EXACTLY ONE source: a raw string constant, a `graphInput` wrapper, or a `taskOutput` wrapper (mixing sources in one object is rejected). The `graphInput` / `taskOutput` wrappers use the same shapes as runnable Tangle pipelines. The ONLY dehydrated-vs-runnable difference is `componentRef` (a lightweight ref) vs an inline `componentSpec`; argument values are identical to the runnable contract.", + "description": "Source of a task argument or graph output. Each argument has EXACTLY ONE source: a raw string constant, a `graphInput` wrapper, a `taskOutput` wrapper, or a `dynamicData` wrapper (mixing sources in one object is rejected). The `graphInput` / `taskOutput` / `dynamicData` wrappers use the same shapes as runnable Tangle pipelines. The ONLY dehydrated-vs-runnable difference is `componentRef` (a lightweight ref) vs an inline `componentSpec`; argument values are identical to the runnable contract.", "oneOf": [ { "type": "string" }, { "$ref": "#/$defs/GraphInputArgument" }, - { "$ref": "#/$defs/TaskOutputArgument" } + { "$ref": "#/$defs/TaskOutputArgument" }, + { "$ref": "#/$defs/DynamicDataArgument" } ] }, @@ -221,6 +222,33 @@ } }, + "DynamicDataArgument": { + "type": "object", + "additionalProperties": false, + "required": ["dynamicData"], + "properties": { + "dynamicData": { "$ref": "#/$defs/DynamicSecretData" } + } + }, + + "DynamicSecretData": { + "type": "object", + "additionalProperties": false, + "required": ["secret"], + "properties": { + "secret": { "$ref": "#/$defs/DynamicSecretReference" } + } + }, + + "DynamicSecretReference": { + "type": "object", + "additionalProperties": false, + "required": ["name"], + "properties": { + "name": { "type": "string", "minLength": 1 } + } + }, + "ExecutionOptionsSpec": { "type": "object", "additionalProperties": true, diff --git a/tests/fixtures/python_pipeline/dynamic_secret_pipeline.py b/tests/fixtures/python_pipeline/dynamic_secret_pipeline.py new file mode 100644 index 0000000..fa496aa --- /dev/null +++ b/tests/fixtures/python_pipeline/dynamic_secret_pipeline.py @@ -0,0 +1,10 @@ +"""Minimal compile fixture for a task argument sourced from a runtime secret.""" + +from tangle_cli.python_pipeline import dynamic_secret, pipeline, ref + +MODEL = ref(name="model") + + +@pipeline(name="Dynamic Secret") +def dynamic_secret_pipeline() -> None: + MODEL.named("Call Model")(openai_api_key=dynamic_secret("OPENAI_API_KEY")) diff --git a/tests/test_python_pipeline_dsl.py b/tests/test_python_pipeline_dsl.py index b693ab7..3a876e9 100644 --- a/tests/test_python_pipeline_dsl.py +++ b/tests/test_python_pipeline_dsl.py @@ -56,6 +56,7 @@ def test_all_names_are_exported(self): "registered", "ref", "raw", + "dynamic_secret", "subpipeline", "TaskEnv", "In", diff --git a/tests/test_python_pipeline_dynamic_data.py b/tests/test_python_pipeline_dynamic_data.py new file mode 100644 index 0000000..76319c4 --- /dev/null +++ b/tests/test_python_pipeline_dynamic_data.py @@ -0,0 +1,79 @@ +"""Tests for ``dynamicData`` argument wrappers in Python-authored pipelines. + +The dehydrated-pipeline schema is deliberately strict: a ``dynamicData`` +argument must be a non-empty ``secret.name`` reference and nothing else. +(The runnable ``pipeline_schema.json`` stays intentionally permissive — it +mirrors the real Tangle runtime, which resolves arbitrary dynamic data such +as run IDs and loop indices — so strictness is enforced here, at compile / +dehydrated-validation time.) +""" +import copy +from pathlib import Path + +import pytest +import yaml + +from tangle_cli.pipeline_compiler import compile_pipeline +from tangle_cli.python_pipeline.dynamic_data import dynamic_secret +from tangle_cli.python_pipeline.emit import emit_pipeline +from tangle_cli.python_pipeline.graph import GraphBuilder, TaskNode +from tangle_cli.schema_validation import SchemaValidationError, validate_dehydrated_pipeline + +FIXTURES = Path(__file__).parent / "fixtures" / "python_pipeline" + + +def _body_with_dynamic_secret() -> dict: + builder = GraphBuilder(name="Demo") + builder.add_task( + TaskNode( + task_id="Call Model", + ref_url="file://./noop.yaml", + arguments={"openai_api_key": dynamic_secret("OPENAI_API_KEY")}, + ) + ) + return emit_pipeline(builder)[0] + + +def test_dynamic_secret_emits_dynamic_data_argument(): + body = _body_with_dynamic_secret() + + emitted = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"]["openai_api_key"] + assert emitted == {"dynamicData": {"secret": {"name": "OPENAI_API_KEY"}}} + validate_dehydrated_pipeline(body) + + +def test_dynamic_secret_fixture_compiles_and_validates(tmp_path): + """Keep a real Python pipeline example alongside the dynamic-data tests.""" + output = tmp_path / "dynamic_secret.yaml" + compile_pipeline(FIXTURES / "dynamic_secret_pipeline.py", output) + body = yaml.safe_load(output.read_text()) + + argument = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"] + assert argument["openai_api_key"] == { + "dynamicData": {"secret": {"name": "OPENAI_API_KEY"}} + } + validate_dehydrated_pipeline(body) + + +@pytest.mark.parametrize( + "dynamic_data", + [ + {}, + {"secret": {}}, + {"secret": {"name": ""}}, + {"arbitrary": [1]}, + {"secret": {"name": "OPENAI_API_KEY", "extra": "value"}}, + ], +) +def test_dynamic_data_rejects_invalid_secret_references(dynamic_data: dict): + body = _body_with_dynamic_secret() + argument = body["implementation"]["graph"]["tasks"]["Call Model"]["arguments"]["openai_api_key"] + argument["dynamicData"] = copy.deepcopy(dynamic_data) + + with pytest.raises(SchemaValidationError): + validate_dehydrated_pipeline(body) + + +def test_dynamic_secret_requires_non_empty_name(): + with pytest.raises(ValueError, match="non-empty secret name"): + dynamic_secret("")