diff --git a/README.md b/README.md index 7160d5c..600f291 100644 --- a/README.md +++ b/README.md @@ -1303,6 +1303,26 @@ def greeting_pipeline() -> Out[str]: Use `unwrap=["items", "metadata"]` to unwrap multiple dict parameters. The caller owns the key names; keys may contain letters, numbers, `_`, and `-`, and become `param__` component inputs. Empty dicts are rejected because they do not define a component interface. If a generated name would collide with a fixed parameter or another generated name, compile fails before writing artifacts. Equivalent key sets are canonicalized for schema hashing, so two call sites with the same keys in different insertion orders dedupe to the same component fragment. +Unwrapped values can also be artifacts. Annotate the dict values with `components.InputPath(...)` (use the builtin `dict[...]`; `typing.Dict` rejects non-type arguments) and every key becomes an `inputPath` input of that type. At runtime the function receives a dict mapping each key to the local path of its downloaded artifact: + +```python +@task(image="python:3.12", unwrap="tables") +def merge_parquets( + tables: dict[str, components.InputPath("ApacheParquet")], + merged_path: components.OutputPath("ApacheParquet"), +): + import pyarrow as pa + import pyarrow.parquet as pq + + pq.write_table(pa.concat_tables([pq.read_table(tables[key]) for key in sorted(tables)]), merged_path) + + +# In a pipeline: any number of upstream Parquet outputs, keyed by the caller. +merged = merge_parquets.named("merge")(tables={"shop": shop.table, "catalog": catalog.table}) +``` + +Artifact unwraps persist `"kind": "input_path"` on each key (and `"value_kind": "input_path"` on the parameter) in `local_from_python.unwrapped_inputs`, so hydrate regenerates the same `inputPath` interface. Value unwraps omit these fields; entries without them are value inputs. `OutputPath` dict values are rejected. + #### Pipeline run submission and validation `submit` hydrates refs by default and builds an API submit payload with `root_task.componentRef.spec`. Use `--no-hydrate` to submit the local YAML structure as-is. Use `--dry-run` to print the payload without creating a run. diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index 6b837cb..c7e0852 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.32" + __version__ = "0.1.33" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/component_from_func.py b/packages/tangle-cli/src/tangle_cli/component_from_func.py index 6355a74..112439c 100644 --- a/packages/tangle-cli/src/tangle_cli/component_from_func.py +++ b/packages/tangle-cli/src/tangle_cli/component_from_func.py @@ -345,30 +345,41 @@ def _dict_value_annotation(annotation: Any) -> Any: return None -def _unwrapped_value_type(annotation: Any) -> tuple[str, str]: - """Infer the Tangle scalar type/deserializer for unwrapped dict values. +_UNWRAPPED_KINDS = ("input", "input_path") + + +def _unwrapped_value_spec(annotation: Any) -> tuple[str, str, Literal["input", "input_path"]]: + """Infer the Tangle type, deserializer and kind for unwrapped dict values. Args: annotation: The resolved annotation for the original dict parameter. Returns: - ``(tangle_type, deserializer)`` for each synthetic flattened input. - Unknown ``Any`` values default to ``String``/``str`` to match existing + ``(tangle_type, deserializer, kind)`` for each synthetic flattened + input. ``kind`` is ``"input"`` for value inputs (``inputValue`` + placeholders) and ``"input_path"`` when the dict values are annotated + as ``InputPath(...)`` artifacts (``inputPath`` placeholders; the + runtime receives one local file path per key). Unknown ``Any`` values + default to ``String``/``str`` value inputs to match existing unannotated input behavior. Raises: ValueError: If the unwrap target is not annotated as ``dict`` or - ``dict[..., ...]``. + ``dict[..., ...]``, or its values are ``OutputPath`` annotations. """ value_annotation = _dict_value_annotation(annotation) if value_annotation is None: raise ValueError("@task unwrap parameters must be annotated as dict[...] or dict") if value_annotation is Any: - return "String", "str" + return "String", "str", "input" tangle_type, deserializer, kind = _resolve_annotation(value_annotation) - if kind != "input" or not tangle_type: - return "String", "str" - return tangle_type, deserializer + if kind == "input_path": + return tangle_type or "String", "str", "input_path" + if kind == "output": + raise ValueError("@task unwrap parameters cannot have OutputPath values; only inputs can be unwrapped") + if not tangle_type: + return "String", "str", "input" + return tangle_type, deserializer, "input" def _normalize_unwrapped_input_keys( @@ -435,7 +446,11 @@ def build_unwrapped_inputs_schema( writes this into the resolver sidecar so hydrate can regenerate the identical component interface without access to the original call-site dict. Each key entry records the original key, synthetic input name, - inferred Tangle type, and optional flag. + inferred Tangle type, and optional flag. Artifact unwraps + (``dict[str, InputPath(...)]``) additionally record + ``"kind": "input_path"`` on each key and ``"value_kind": + "input_path"`` on the parameter entry; value unwraps omit both keys so + their persisted schema (and fragment hash) is unchanged. Raises: ValueError: If an unwrap parameter is absent from the function @@ -464,7 +479,7 @@ def build_unwrapped_inputs_schema( if param_name not in signature.parameters: raise ValueError(f"@task unwrap parameter {param_name!r} is not in the function signature") annotation = resolved_hints.get(param_name, signature.parameters[param_name].annotation) - tangle_type, _deserializer = _unwrapped_value_type(annotation) + tangle_type, _deserializer, kind = _unwrapped_value_spec(annotation) input_prefix = f"{param_name}__" key_specs = [] for key in keys: @@ -480,19 +495,23 @@ def build_unwrapped_inputs_schema( f"which collides with unwrap parameter {seen_synthetic_names[input_name]!r}" ) seen_synthetic_names[input_name] = param_name - key_specs.append( - { - "key": key, - "input_name": input_name, - "type": tangle_type, - "optional": False, - } - ) - schema[param_name] = { + key_spec: dict[str, Any] = { + "key": key, + "input_name": input_name, + "type": tangle_type, + "optional": False, + } + if kind != "input": + key_spec["kind"] = kind + key_specs.append(key_spec) + param_schema: dict[str, Any] = { "input_prefix": input_prefix, "value_type": tangle_type, - "keys": key_specs, } + if kind != "input": + param_schema["value_kind"] = kind + param_schema["keys"] = key_specs + schema[param_name] = param_schema return schema @@ -517,28 +536,40 @@ def _expand_unwrapped_param( A ``ParamInfo`` for each flattened key, using names like ``run_data__shop``. The returned params retain ``source_param`` and ``source_key`` so generated runtime code can re-wrap CLI arguments into - the original dict before invoking the function. + the original dict before invoking the function. The persisted + ``kind``/``value_kind`` (default ``"input"``) decides whether each + key becomes an ``inputValue`` or an ``inputPath`` input; artifact + inputs always receive the local path as a plain ``str``. Raises: - ValueError: If ``annotation`` is not dict-like. + ValueError: If ``annotation`` is not dict-like, or the schema records + an unknown kind. """ - _unwrapped_value_type(annotation) # validates dict-like annotation + _unwrapped_value_spec(annotation) # validates dict-like annotation + default_kind = schema.get("value_kind") or "input" expanded: list[ParamInfo] = [] for key_spec in schema.get("keys", []) or []: key = str(key_spec["key"]) input_name = str(key_spec.get("input_name") or f"{param.name}__{key}") tangle_type = str(key_spec.get("type") or schema.get("value_type") or "String") + kind = key_spec.get("kind") or default_kind + if kind not in _UNWRAPPED_KINDS: + raise ValueError( + f"@task unwrap parameter {param.name!r} key {key!r} has unsupported kind {kind!r}; " + f"expected one of {list(_UNWRAPPED_KINDS)}" + ) + deserializer = "str" if kind == "input_path" else _TYPE_TO_DESERIALIZER.get(tangle_type, "str") expanded.append( ParamInfo( name=input_name, yaml_name=input_name, python_type=str(annotation) if annotation is not inspect.Parameter.empty else None, tangle_type=tangle_type, - kind="input", + kind=kind, description=doc_dict.get(param.name), default=inspect.Parameter.empty, optional=bool(key_spec.get("optional", False)), - deserializer=_TYPE_TO_DESERIALIZER.get(tangle_type, "str"), + deserializer=deserializer, source_param=param.name, source_key=key, ) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/task.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/task.py index b52561d..b595c0d 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/task.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/task.py @@ -123,7 +123,10 @@ def task( For example ``unwrap="run_data"`` turns ``run_data={"run_id_1": task.output}`` into a component input named ``run_data__run_id_1`` and reconstructs the original dict in - the generated runtime wrapper. + the generated runtime wrapper. Values annotated as + ``dict[str, components.InputPath(T)]`` become ``inputPath`` + artifact inputs, and the reconstructed dict maps each key to the + artifact's local path. Returns: A decorator that, given the user's function, returns a diff --git a/pyproject.toml b/pyproject.toml index 6dde619..4de5b27 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.32" +version = "0.1.33" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_component_from_func.py b/tests/test_component_from_func.py index 9cd0806..6e1ec74 100644 --- a/tests/test_component_from_func.py +++ b/tests/test_component_from_func.py @@ -93,6 +93,101 @@ def uses_int_values(metrics: dict[str, int]) -> str: assert [item["type"] for item in schema["metrics"]["keys"]] == ["Integer", "Integer"] +def test_unwrapped_value_schema_has_no_kind_keys(): + """Value unwraps keep the historical schema shape (and fragment hash).""" + from typing import Dict + + def uses_json_values(run_metrics: Dict[str, dict]) -> str: # build_metrics_dict style + return "ok" + + schema = cff.build_unwrapped_inputs_schema(uses_json_values, {"run_metrics": ["b", "a"]}) + + assert schema == { + "run_metrics": { + "input_prefix": "run_metrics__", + "value_type": "JsonObject", + "keys": [ + {"key": "a", "input_name": "run_metrics__a", "type": "JsonObject", "optional": False}, + {"key": "b", "input_name": "run_metrics__b", "type": "JsonObject", "optional": False}, + ], + } + } + spec = extract_interface(uses_json_values, {}, unwrapped_inputs=schema) + assert [(p.yaml_name, p.kind, p.deserializer) for p in spec.inputs] == [ + ("run_metrics__a", "input", "json.loads"), + ("run_metrics__b", "input", "json.loads"), + ] + + +def test_unwrapped_input_path_schema_records_kind(): + def merge(tables: dict[str, InputPath("ApacheParquet")]) -> str: + return "ok" + + schema = cff.build_unwrapped_inputs_schema(merge, {"tables": ["shop", "catalog"]}) + + assert schema == { + "tables": { + "input_prefix": "tables__", + "value_type": "ApacheParquet", + "value_kind": "input_path", + "keys": [ + { + "key": "catalog", + "input_name": "tables__catalog", + "type": "ApacheParquet", + "optional": False, + "kind": "input_path", + }, + { + "key": "shop", + "input_name": "tables__shop", + "type": "ApacheParquet", + "optional": False, + "kind": "input_path", + }, + ], + } + } + spec = extract_interface(merge, {}, unwrapped_inputs=schema) + assert [(p.yaml_name, p.kind, p.tangle_type, p.deserializer) for p in spec.inputs] == [ + ("tables__catalog", "input_path", "ApacheParquet", "str"), + ("tables__shop", "input_path", "ApacheParquet", "str"), + ] + + +def test_unwrapped_untyped_input_path_defaults_to_string_artifact(): + def merge(files: dict[str, InputPath()]) -> str: + return "ok" + + schema = cff.build_unwrapped_inputs_schema(merge, {"files": ["a"]}) + + assert schema["files"]["value_kind"] == "input_path" + assert schema["files"]["keys"][0]["type"] == "String" + + +def test_unwrapped_output_path_values_are_rejected(): + def bad(outs: dict[str, OutputPath("Text")]) -> str: + return "ok" + + with pytest.raises(ValueError, match="cannot have OutputPath values"): + cff.build_unwrapped_inputs_schema(bad, {"outs": ["a"]}) + + +def test_unwrapped_schema_with_unknown_kind_is_rejected(): + def merge(tables: dict[str, str]) -> str: + return "ok" + + schema = { + "tables": { + "input_prefix": "tables__", + "value_type": "String", + "keys": [{"key": "a", "input_name": "tables__a", "type": "String", "kind": "output"}], + } + } + with pytest.raises(ValueError, match="unsupported kind 'output'"): + extract_interface(merge, {}, unwrapped_inputs=schema) + + # ============================================================================ # Type resolution tests # ============================================================================ @@ -3039,3 +3134,142 @@ def greet(name: str) -> str: args = component["implementation"]["container"]["args"] assert "----output-paths" in args assert {"outputPath": "Output"} in args + + +def test_generate_component_yaml_with_unwrapped_input_paths_runs_rewrapped_program(tmp_path): + source = tmp_path / "merge_component.py" + source.write_text( + textwrap.dedent( + ''' + from cloud_pipelines import components + + def merge_component( + out: components.OutputPath("Text"), + tables: dict[str, components.InputPath("ApacheParquet")], + ): + parts = [] + for key in sorted(tables): + with open(tables[key]) as fh: + parts.append(key + "=" + fh.read()) + with open(out, "w") as fh: + fh.write("|".join(parts)) + ''' + ).lstrip() + ) + module = load_python_module(source) + func = get_function_from_module(module, "merge_component") + unwrapped_inputs = cff.build_unwrapped_inputs_schema(func, {"tables": ["shop", "catalog"]}) + output_file = tmp_path / "component.yaml" + + assert generate_component_yaml( + source, + output_file, + container_image="python:3.12", + function_name="merge_component", + unwrapped_inputs=unwrapped_inputs, + ) is True + + component = yaml.safe_load(output_file.read_text()) + assert [(item["name"], item["type"]) for item in component["inputs"]] == [ + ("tables__catalog", "ApacheParquet"), + ("tables__shop", "ApacheParquet"), + ] + args = component["implementation"]["container"]["args"] + assert {"inputPath": "tables__catalog"} in args + assert {"inputPath": "tables__shop"} in args + assert {"inputValue": "tables__catalog"} not in args + + shop = tmp_path / "shop.parquet" + shop.write_text("S") + catalog = tmp_path / "catalog.parquet" + catalog.write_text("C") + program = component["implementation"]["container"]["command"][-1] + program_path = tmp_path / "program.py" + program_path.write_text(program) + out_path = tmp_path / "merged.txt" + completed = subprocess.run( + [ + sys.executable, + str(program_path), + "--out", + str(out_path), + "--tables__catalog", + str(catalog), + "--tables__shop", + str(shop), + ], + capture_output=True, + check=False, + text=True, + ) + assert completed.returncode == 0, completed.stderr + assert out_path.read_text() == "catalog=C|shop=S" + + +def test_generate_component_yaml_with_multiple_mixed_unwraps_runs_rewrapped_program(tmp_path): + """Two artifact unwraps plus a value unwrap on one task, with hyphenated keys.""" + source = tmp_path / "multi_component.py" + source.write_text( + textwrap.dedent( + ''' + from cloud_pipelines import components + + def multi_component( + out: components.OutputPath("Text"), + raw: dict[str, components.InputPath("ApacheParquet")], + results: dict[str, components.InputPath("ApacheParquet")], + labels: dict[str, str], + ): + parts = [] + for name, group in (("raw", raw), ("results", results)): + for key in sorted(group): + with open(group[key]) as fh: + parts.append(name + ":" + key + "=" + fh.read() + "/" + labels[key]) + with open(out, "w") as fh: + fh.write("|".join(parts)) + ''' + ).lstrip() + ) + module = load_python_module(source) + func = get_function_from_module(module, "multi_component") + keys = ["ucp-default", "ucp_alt"] + unwrapped_inputs = cff.build_unwrapped_inputs_schema(func, {"raw": keys, "results": keys, "labels": keys}) + assert [unwrapped_inputs[name].get("value_kind") for name in ("raw", "results", "labels")] == [ + "input_path", + "input_path", + None, + ] + output_file = tmp_path / "component.yaml" + + assert generate_component_yaml( + source, + output_file, + container_image="python:3.12", + function_name="multi_component", + unwrapped_inputs=unwrapped_inputs, + ) is True + + component = yaml.safe_load(output_file.read_text()) + args = component["implementation"]["container"]["args"] + for name in ("raw", "results"): + for key in keys: + assert {"inputPath": f"{name}__{key}"} in args + for key in keys: + assert {"inputValue": f"labels__{key}"} in args + + argv = [sys.executable, "PROGRAM", "--out", str(tmp_path / "out.txt")] + for name in ("raw", "results"): + for key in keys: + path = tmp_path / f"{name}-{key}.parquet" + path.write_text(f"{name}.{key}") + argv += [f"--{name}__{key}", str(path)] + argv += ["--labels__ucp-default", "D", "--labels__ucp_alt", "A"] + program_path = tmp_path / "program.py" + program_path.write_text(component["implementation"]["container"]["command"][-1]) + argv[1] = str(program_path) + completed = subprocess.run(argv, capture_output=True, check=False, text=True) + assert completed.returncode == 0, completed.stderr + assert (tmp_path / "out.txt").read_text() == ( + "raw:ucp-default=raw.ucp-default/D|raw:ucp_alt=raw.ucp_alt/A|" + "results:ucp-default=results.ucp-default/D|results:ucp_alt=results.ucp_alt/A" + ) diff --git a/tests/test_packaging.py b/tests/test_packaging.py index bca0db8..08e316c 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.32" in metadata + assert "Version: 0.1.33" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_pipeline_compiler.py b/tests/test_pipeline_compiler.py index 0e86cac..4056969 100644 --- a/tests/test_pipeline_compiler.py +++ b/tests/test_pipeline_compiler.py @@ -447,6 +447,68 @@ def test_compile_task_decorator_emits_unwrapped_input_schema_and_flat_edges(tmp_ ] +def test_compile_task_decorator_unwraps_input_path_artifacts_and_hydrates(tmp_path): + """``dict[str, InputPath(T)]`` unwraps to one ``inputPath`` input per key. + + The persisted schema records the artifact kind, so hydrate regenerates the + same artifact interface from the sidecar alone. + """ + project = tmp_path / "project" + src = project / "src" + pipeline_path = src / "pipeline.py" + src.mkdir(parents=True) + pipeline_path.write_text( + "from cloud_pipelines import components\n" + "from tangle_cli.python_pipeline import Out, pipeline, task\n\n" + "@task(image='python:3.12')\n" + "def produce(table_path: components.OutputPath('ApacheParquet')):\n" + " open(table_path, 'w').close()\n\n" + "@task(image='python:3.12', unwrap='tables')\n" + "def merge(\n" + " tables: dict[str, components.InputPath('ApacheParquet')],\n" + " merged_path: components.OutputPath('ApacheParquet'),\n" + "):\n" + " open(merged_path, 'w').close()\n\n" + "@pipeline('Unwrap Paths Pipeline')\n" + "def unwrap_paths_pipeline() -> Out[str]:\n" + " shop = produce.named('shop')()\n" + " catalog = produce.named('catalog')()\n" + " merged = merge.named('merge')(tables={'shop': shop.table, 'catalog': catalog.table})\n" + " return merged.merged\n", + encoding="utf-8", + ) + + out = project / "compiled.yaml" + result = compile_pipeline(pipeline_path, out) + + compiled = yaml.safe_load(out.read_text()) + merge_task = compiled["implementation"]["graph"]["tasks"]["merge"] + assert set(merge_task["arguments"]) == {"tables__shop", "tables__catalog"} + assert merge_task["arguments"]["tables__shop"]["taskOutput"] == {"taskId": "shop", "outputName": "table"} + + sidecar = yaml.safe_load(result.components_path.read_text()) + fragments = [name for name in sidecar if name.startswith("merge--")] + assert len(fragments) == 1 + schema = sidecar[fragments[0]]["local_from_python"]["unwrapped_inputs"]["tables"] + assert schema["value_type"] == "ApacheParquet" + assert schema["value_kind"] == "input_path" + assert [(key["key"], key["type"], key["kind"]) for key in schema["keys"]] == [ + ("catalog", "ApacheParquet", "input_path"), + ("shop", "ApacheParquet", "input_path"), + ] + + hydrated = PipelineHydrator(client=MagicMock()).hydrate_file(out).data + spec = hydrated["implementation"]["graph"]["tasks"]["merge"]["componentRef"]["spec"] + assert [(item["name"], item["type"]) for item in spec["inputs"]] == [ + ("tables__catalog", "ApacheParquet"), + ("tables__shop", "ApacheParquet"), + ] + args = spec["implementation"]["container"]["args"] + assert {"inputPath": "tables__catalog"} in args + assert {"inputPath": "tables__shop"} in args + assert not any(isinstance(arg, dict) and "inputValue" in arg for arg in args) + + def test_compile_task_decorator_uses_schema_hash_for_unwrapped_fragment_collisions(tmp_path): project = tmp_path / "project" src = project / "src" diff --git a/uv.lock b/uv.lock index 25a212b..62c2baa 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.32" +version = "0.1.33" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },