Skip to content

Commit 714366d

Browse files
committed
Support InputPath artifact values in @task(unwrap=...) (v0.1.33)
`dict[str, components.InputPath(T)]` unwrap targets now expand into one `inputPath` component input per call-site key. The generated runtime wrapper re-assembles the dict with each key's local artifact path, so a single inline task can consume an arbitrary number of upstream artifacts (e.g. merging N Parquet tables). The artifact kind is persisted in `local_from_python.unwrapped_inputs` (`kind` per key, `value_kind` per parameter) so hydrate regenerates the identical interface. Value unwraps omit both fields, keeping their persisted schema and fragment hash byte-identical. `OutputPath` dict values are now rejected instead of silently becoming String inputs.
1 parent fa17dcb commit 714366d

9 files changed

Lines changed: 381 additions & 31 deletions

File tree

‎README.md‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1303,6 +1303,26 @@ def greeting_pipeline() -> Out[str]:
13031303

13041304
Use `unwrap=["items", "metadata"]` to unwrap multiple dict parameters. The caller owns the key names; keys may contain letters, numbers, `_`, and `-`, and become `param__<key>` 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.
13051305

1306+
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:
1307+
1308+
```python
1309+
@task(image="python:3.12", unwrap="tables")
1310+
def merge_parquets(
1311+
tables: dict[str, components.InputPath("ApacheParquet")],
1312+
merged_path: components.OutputPath("ApacheParquet"),
1313+
):
1314+
import pyarrow as pa
1315+
import pyarrow.parquet as pq
1316+
1317+
pq.write_table(pa.concat_tables([pq.read_table(tables[key]) for key in sorted(tables)]), merged_path)
1318+
1319+
1320+
# In a pipeline: any number of upstream Parquet outputs, keyed by the caller.
1321+
merged = merge_parquets.named("merge")(tables={"shop": shop.table, "catalog": catalog.table})
1322+
```
1323+
1324+
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.
1325+
13061326
#### Pipeline run submission and validation
13071327

13081328
`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.

‎packages/tangle-cli/src/tangle_cli/__init__.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,6 @@
1414
try:
1515
__version__ = metadata_version("tangle-cli")
1616
except PackageNotFoundError:
17-
__version__ = "0.1.32"
17+
__version__ = "0.1.33"
1818

1919
__all__ = ["TangleDynamicDiscoveryClient", "__version__"]

‎packages/tangle-cli/src/tangle_cli/component_from_func.py‎

Lines changed: 57 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -345,30 +345,41 @@ def _dict_value_annotation(annotation: Any) -> Any:
345345
return None
346346

347347

348-
def _unwrapped_value_type(annotation: Any) -> tuple[str, str]:
349-
"""Infer the Tangle scalar type/deserializer for unwrapped dict values.
348+
_UNWRAPPED_KINDS = ("input", "input_path")
349+
350+
351+
def _unwrapped_value_spec(annotation: Any) -> tuple[str, str, Literal["input", "input_path"]]:
352+
"""Infer the Tangle type, deserializer and kind for unwrapped dict values.
350353
351354
Args:
352355
annotation: The resolved annotation for the original dict parameter.
353356
354357
Returns:
355-
``(tangle_type, deserializer)`` for each synthetic flattened input.
356-
Unknown ``Any`` values default to ``String``/``str`` to match existing
358+
``(tangle_type, deserializer, kind)`` for each synthetic flattened
359+
input. ``kind`` is ``"input"`` for value inputs (``inputValue``
360+
placeholders) and ``"input_path"`` when the dict values are annotated
361+
as ``InputPath(...)`` artifacts (``inputPath`` placeholders; the
362+
runtime receives one local file path per key). Unknown ``Any`` values
363+
default to ``String``/``str`` value inputs to match existing
357364
unannotated input behavior.
358365
359366
Raises:
360367
ValueError: If the unwrap target is not annotated as ``dict`` or
361-
``dict[..., ...]``.
368+
``dict[..., ...]``, or its values are ``OutputPath`` annotations.
362369
"""
363370
value_annotation = _dict_value_annotation(annotation)
364371
if value_annotation is None:
365372
raise ValueError("@task unwrap parameters must be annotated as dict[...] or dict")
366373
if value_annotation is Any:
367-
return "String", "str"
374+
return "String", "str", "input"
368375
tangle_type, deserializer, kind = _resolve_annotation(value_annotation)
369-
if kind != "input" or not tangle_type:
370-
return "String", "str"
371-
return tangle_type, deserializer
376+
if kind == "input_path":
377+
return tangle_type or "String", "str", "input_path"
378+
if kind == "output":
379+
raise ValueError("@task unwrap parameters cannot have OutputPath values; only inputs can be unwrapped")
380+
if not tangle_type:
381+
return "String", "str", "input"
382+
return tangle_type, deserializer, "input"
372383

373384

374385
def _normalize_unwrapped_input_keys(
@@ -435,7 +446,11 @@ def build_unwrapped_inputs_schema(
435446
writes this into the resolver sidecar so hydrate can regenerate the
436447
identical component interface without access to the original call-site
437448
dict. Each key entry records the original key, synthetic input name,
438-
inferred Tangle type, and optional flag.
449+
inferred Tangle type, and optional flag. Artifact unwraps
450+
(``dict[str, InputPath(...)]``) additionally record
451+
``"kind": "input_path"`` on each key and ``"value_kind":
452+
"input_path"`` on the parameter entry; value unwraps omit both keys so
453+
their persisted schema (and fragment hash) is unchanged.
439454
440455
Raises:
441456
ValueError: If an unwrap parameter is absent from the function
@@ -464,7 +479,7 @@ def build_unwrapped_inputs_schema(
464479
if param_name not in signature.parameters:
465480
raise ValueError(f"@task unwrap parameter {param_name!r} is not in the function signature")
466481
annotation = resolved_hints.get(param_name, signature.parameters[param_name].annotation)
467-
tangle_type, _deserializer = _unwrapped_value_type(annotation)
482+
tangle_type, _deserializer, kind = _unwrapped_value_spec(annotation)
468483
input_prefix = f"{param_name}__"
469484
key_specs = []
470485
for key in keys:
@@ -480,19 +495,23 @@ def build_unwrapped_inputs_schema(
480495
f"which collides with unwrap parameter {seen_synthetic_names[input_name]!r}"
481496
)
482497
seen_synthetic_names[input_name] = param_name
483-
key_specs.append(
484-
{
485-
"key": key,
486-
"input_name": input_name,
487-
"type": tangle_type,
488-
"optional": False,
489-
}
490-
)
491-
schema[param_name] = {
498+
key_spec: dict[str, Any] = {
499+
"key": key,
500+
"input_name": input_name,
501+
"type": tangle_type,
502+
"optional": False,
503+
}
504+
if kind != "input":
505+
key_spec["kind"] = kind
506+
key_specs.append(key_spec)
507+
param_schema: dict[str, Any] = {
492508
"input_prefix": input_prefix,
493509
"value_type": tangle_type,
494-
"keys": key_specs,
495510
}
511+
if kind != "input":
512+
param_schema["value_kind"] = kind
513+
param_schema["keys"] = key_specs
514+
schema[param_name] = param_schema
496515
return schema
497516

498517

@@ -517,28 +536,40 @@ def _expand_unwrapped_param(
517536
A ``ParamInfo`` for each flattened key, using names like
518537
``run_data__shop``. The returned params retain ``source_param`` and
519538
``source_key`` so generated runtime code can re-wrap CLI arguments into
520-
the original dict before invoking the function.
539+
the original dict before invoking the function. The persisted
540+
``kind``/``value_kind`` (default ``"input"``) decides whether each
541+
key becomes an ``inputValue`` or an ``inputPath`` input; artifact
542+
inputs always receive the local path as a plain ``str``.
521543
522544
Raises:
523-
ValueError: If ``annotation`` is not dict-like.
545+
ValueError: If ``annotation`` is not dict-like, or the schema records
546+
an unknown kind.
524547
"""
525-
_unwrapped_value_type(annotation) # validates dict-like annotation
548+
_unwrapped_value_spec(annotation) # validates dict-like annotation
549+
default_kind = schema.get("value_kind") or "input"
526550
expanded: list[ParamInfo] = []
527551
for key_spec in schema.get("keys", []) or []:
528552
key = str(key_spec["key"])
529553
input_name = str(key_spec.get("input_name") or f"{param.name}__{key}")
530554
tangle_type = str(key_spec.get("type") or schema.get("value_type") or "String")
555+
kind = key_spec.get("kind") or default_kind
556+
if kind not in _UNWRAPPED_KINDS:
557+
raise ValueError(
558+
f"@task unwrap parameter {param.name!r} key {key!r} has unsupported kind {kind!r}; "
559+
f"expected one of {list(_UNWRAPPED_KINDS)}"
560+
)
561+
deserializer = "str" if kind == "input_path" else _TYPE_TO_DESERIALIZER.get(tangle_type, "str")
531562
expanded.append(
532563
ParamInfo(
533564
name=input_name,
534565
yaml_name=input_name,
535566
python_type=str(annotation) if annotation is not inspect.Parameter.empty else None,
536567
tangle_type=tangle_type,
537-
kind="input",
568+
kind=kind,
538569
description=doc_dict.get(param.name),
539570
default=inspect.Parameter.empty,
540571
optional=bool(key_spec.get("optional", False)),
541-
deserializer=_TYPE_TO_DESERIALIZER.get(tangle_type, "str"),
572+
deserializer=deserializer,
542573
source_param=param.name,
543574
source_key=key,
544575
)

‎packages/tangle-cli/src/tangle_cli/python_pipeline/task.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,10 @@ def task(
123123
For example ``unwrap="run_data"`` turns
124124
``run_data={"run_id_1": task.output}`` into a component input
125125
named ``run_data__run_id_1`` and reconstructs the original dict in
126-
the generated runtime wrapper.
126+
the generated runtime wrapper. Values annotated as
127+
``dict[str, components.InputPath(T)]`` become ``inputPath``
128+
artifact inputs, and the reconstructed dict maps each key to the
129+
artifact's local path.
127130
128131
Returns:
129132
A decorator that, given the user's function, returns a

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
[project]
22
name = "tangle-cli"
3-
version = "0.1.32"
3+
version = "0.1.33"
44
description = "CLI for Tangle, the open-source ML pipeline orchestration platform"
55
readme = "README.md"
66
authors = [

0 commit comments

Comments
 (0)