diff --git a/CHANGELOG.md b/CHANGELOG.md index f32c9fc7..98262cf6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ - `CacheDict`: deletions require a `write()` context (like writes). [#326] - `DumpContext.shared_file`: content suffixes must start with `.`. [#326] +- `steps`: `Parallel` rejects input from an upstream step. [#328] [other] diff --git a/exca/steps/helpers.py b/exca/steps/helpers.py index 3d68939c..6db39e3d 100644 --- a/exca/steps/helpers.py +++ b/exca/steps/helpers.py @@ -128,7 +128,8 @@ class Parallel(Step): The variants run together under one shared backend, each caching under its own identity. ``run`` is for effect — read results back per variant via ``parallel.steps[k].lookup(value)``. It has no composable output (yields - ``None`` per input), so use it standalone, not as a non-terminal chain step. + ``None`` per input), so it cannot consume another step's output — run it + standalone. Example:: @@ -179,9 +180,6 @@ def _unify_infra(self) -> None: f"steps; {self.infra!r} differs from {step.infra!r}" ) - def _uid_steps(self) -> list[Step]: - return [] # no identity of its own - def lookup(self, *args: tp.Any, **kwargs: tp.Any) -> tp.NoReturn: raise TypeError( "Parallel has no cache of its own; look up a variant instead, " @@ -189,6 +187,11 @@ def lookup(self, *args: tp.Any, **kwargs: tp.Any) -> tp.NoReturn: ) def _dispatch(self, batch: items.StepItems) -> items.StepItems: + if batch._upstream: + raise TypeError( + "Parallel has no output to pass on, so it cannot consume another " + "step's output; run it standalone." + ) return self._run_items(batch) # not infra._run(self): dispatch variants def _run_items(self, batch: items.StepItems) -> items.StepItems: diff --git a/exca/steps/test_helpers.py b/exca/steps/test_helpers.py index 489f6634..efdfab32 100644 --- a/exca/steps/test_helpers.py +++ b/exca/steps/test_helpers.py @@ -142,6 +142,11 @@ def test_run_caches_each_variant_under_own_identity(tmp_path: Path) -> None: assert [s.lookup(5.0).result() for s in sweep.steps] == [10.0, 15.0, 20.0] infra: tp.Any = {"backend": "Cached", "folder": tmp_path} assert conftest.Mult(coeff=3.0, infra=infra).lookup(5.0).cached() + Chain(steps=[_sweep(tmp_path)], infra=infra).run_many([1.0]) # allowed: no upstream + folders = conftest.extract_cache_folders(tmp_path) + assert any(f.startswith("type=Parallel") for f in folders), ( + f"chain-leading Parallel must cache under its own identity, got {folders}" + ) def test_generator_variants_no_items(tmp_path: Path) -> None: @@ -163,6 +168,8 @@ def test_invalid_inputs_rejected(tmp_path: Path) -> None: Parallel(steps=conflicting, infra=infra) with pytest.raises(TypeError, match="parallel.steps"): _sweep(tmp_path).lookup(5.0) + with pytest.raises(TypeError, match="cannot consume another step's output"): + Chain(steps=[conftest.Add(value=100.0), _sweep(tmp_path)]).run_many([3.0]) @pytest.mark.parametrize("folder_first", (True, False))