From 180a000c58f273822bed45a01e1598db9b5776b1 Mon Sep 17 00:00:00 2001 From: Jeremy Rapin Date: Tue, 22 Sep 2026 16:32:10 +0200 Subject: [PATCH 1/3] Prevent Parallel from being in a Chain --- CHANGELOG.md | 1 + exca/steps/helpers.py | 10 ++++++---- exca/steps/test_helpers.py | 7 +++++++ 3 files changed, 14 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f32c9fc7..9bb205a7 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` can no longer be a `Chain` step. [#NNN] [other] diff --git a/exca/steps/helpers.py b/exca/steps/helpers.py index 3d68939c..45fa58d5 100644 --- a/exca/steps/helpers.py +++ b/exca/steps/helpers.py @@ -128,7 +128,7 @@ 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 be a ``Chain`` step — run it standalone. Example:: @@ -179,9 +179,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 +186,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 be a Chain step; " + "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..bbcfb36d 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 be a Chain step"): + Chain(steps=[conftest.Add(value=100.0), _sweep(tmp_path)]).run_many([3.0]) @pytest.mark.parametrize("folder_first", (True, False)) From 6b8d69cdc18f755007b8dc9210f52cd912aa176d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=A9r=C3=A9my=20Rapin?= Date: Tue, 22 Sep 2026 16:32:58 +0200 Subject: [PATCH 2/3] Update CHANGELOG.md --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bb205a7..15a738e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ - `CacheDict`: deletions require a `write()` context (like writes). [#326] - `DumpContext.shared_file`: content suffixes must start with `.`. [#326] -- `steps`: `Parallel` can no longer be a `Chain` step. [#NNN] +- `steps`: `Parallel` can no longer be a `Chain` step. [#328] [other] From 433c8d2cefb1aa7607dc5b011b524860e62ac6ff Mon Sep 17 00:00:00 2001 From: Jeremy Rapin Date: Tue, 22 Sep 2026 16:38:56 +0200 Subject: [PATCH 3/3] fix --- CHANGELOG.md | 2 +- exca/steps/helpers.py | 7 ++++--- exca/steps/test_helpers.py | 2 +- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9bb205a7..98262cf6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ - `CacheDict`: deletions require a `write()` context (like writes). [#326] - `DumpContext.shared_file`: content suffixes must start with `.`. [#326] -- `steps`: `Parallel` can no longer be a `Chain` step. [#NNN] +- `steps`: `Parallel` rejects input from an upstream step. [#328] [other] diff --git a/exca/steps/helpers.py b/exca/steps/helpers.py index 45fa58d5..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 it cannot be a ``Chain`` step — run it standalone. + ``None`` per input), so it cannot consume another step's output — run it + standalone. Example:: @@ -188,8 +189,8 @@ 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 be a Chain step; " - "run it standalone." + "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 diff --git a/exca/steps/test_helpers.py b/exca/steps/test_helpers.py index bbcfb36d..efdfab32 100644 --- a/exca/steps/test_helpers.py +++ b/exca/steps/test_helpers.py @@ -168,7 +168,7 @@ 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 be a Chain step"): + 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])