From d76951fe476944348c3c75617feb1618a32905a3 Mon Sep 17 00:00:00 2001 From: Konstantin Fickel Date: Sat, 21 Feb 2026 11:46:07 +0100 Subject: [PATCH] feat: add regenerate command to force rebuild of specific targets The regenerate command accepts one or more target names and forces them to be rebuilt even if they are up to date. This respects the archive_folder setting, archiving previous versions before overwriting. --- hokusai/builder.py | 8 +++++- hokusai/cli.py | 66 ++++++++++++++++++++++++++++++++++++++++++++-- tests/test_cli.py | 53 +++++++++++++++++++++++++++++++++++++ 3 files changed, 124 insertions(+), 3 deletions(-) diff --git a/hokusai/builder.py b/hokusai/builder.py index 8a0f43b..34b6ec5 100644 --- a/hokusai/builder.py +++ b/hokusai/builder.py @@ -179,6 +179,7 @@ async def run_build( project_dir: Path, project_name: str, target: str | None = None, + force_dirty: set[str] | None = None, on_progress: ProgressCallback = _noop_callback, ) -> BuildResult: """Execute the build. @@ -186,10 +187,14 @@ async def run_build( If *target* is specified, only build that target and its transitive dependencies. Otherwise build all targets. + If *force_dirty* is provided, those targets are always considered dirty + regardless of their actual state (used by the ``regenerate`` command). + Execution proceeds in topological generations — each generation is a set of independent targets that run concurrently via :func:`asyncio.gather`. """ + force_dirty = force_dirty or set() result = BuildResult() providers = _create_providers() provider_index = _build_provider_index(providers) @@ -221,7 +226,8 @@ async def run_build( on_progress(BuildEvent.TARGET_DEP_FAILED, name, "Dependency failed") continue - if _is_dirty(name, config, project_dir, state): + is_forced = name in force_dirty + if is_forced or _is_dirty(name, config, project_dir, state): if not _has_provider(name, config, provider_index, result, on_progress): continue dirty_targets.append(name) diff --git a/hokusai/cli.py b/hokusai/cli.py index 5c149a4..455fd2f 100644 --- a/hokusai/cli.py +++ b/hokusai/cli.py @@ -106,7 +106,11 @@ def _format_elapsed(seconds: float) -> str: def _run_build( - config: ProjectConfig, project_dir: Path, project_name: str, target: str | None + config: ProjectConfig, + project_dir: Path, + project_name: str, + target: str | None, + force_dirty: set[str] | None = None, ) -> tuple[BuildResult, float]: """Run the async build with click-styled progress output and timing.""" @@ -142,7 +146,14 @@ def _run_build( start = time.monotonic() result = asyncio.run( - run_build(config, project_dir, project_name, target, on_progress=on_progress) + run_build( + config, + project_dir, + project_name, + target, + force_dirty=force_dirty, + on_progress=on_progress, + ) ) elapsed = time.monotonic() - start @@ -186,6 +197,57 @@ def build( raise typer.Exit(code=1) +@app.command() +def regenerate( + targets: Annotated[list[str], typer.Argument(help="Targets to regenerate.")], + project: Annotated[ + str | None, typer.Option(help="Project name (loads .hokusai.yaml).") + ] = None, +) -> None: + """Force regeneration of specific targets, ignoring up-to-date status.""" + project_dir = Path.cwd() + config_path = _find_config(project_dir, project) + config = load_config(config_path) + name = _project_name(config_path) + + # Validate all targets exist. + unknown = [t for t in targets if t not in config.targets] + if unknown: + click.echo( + click.style("Error: ", fg="red", bold=True) + + f"Unknown target(s): {', '.join(unknown)}", + err=True, + ) + raise typer.Exit(code=1) + + click.echo( + click.style("hokusai", fg="cyan", bold=True) + " regenerating targets...\n" + ) + + result, elapsed = _run_build( + config, project_dir, name, target=None, force_dirty=set(targets) + ) + + # Summary + click.echo("") + parts: list[str] = [] + if result.built: + parts.append( + click.style(f"{len(result.built)} regenerated", fg="green", bold=True) + ) + if result.skipped: + parts.append(click.style(f"{len(result.skipped)} skipped", fg="yellow")) + if result.failed: + parts.append(click.style(f"{len(result.failed)} failed", fg="red", bold=True)) + + summary = ", ".join(parts) if parts else click.style("nothing to do", dim=True) + timer = click.style(f" in {_format_elapsed(elapsed)}", dim=True) + click.echo(summary + timer) + + if result.failed: + raise typer.Exit(code=1) + + @app.command() def clean() -> None: """Remove generated artifacts (targets only, not input files).""" diff --git a/tests/test_cli.py b/tests/test_cli.py index ff4cfb1..4eb1291 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -136,6 +136,59 @@ class TestBuildCommand: assert call_args[0][3] == "output.txt" +class TestRegenerateCommand: + """Test the regenerate CLI command.""" + + def test_regenerate_forces_rebuild(self, cli_project: Path) -> None: + build_result = BuildResult( + built=["output.txt"], skipped=["image.png"], failed={} + ) + with ( + patch("hokusai.cli.Path") as mock_path_cls, + patch( + "hokusai.cli.run_build", + new_callable=AsyncMock, + return_value=build_result, + ) as mock_run, + ): + mock_path_cls.cwd.return_value = cli_project + result = runner.invoke(app, ["regenerate", "output.txt"]) + + assert result.exit_code == 0 + assert "1 regenerated" in result.output + # Check force_dirty was passed + call_kwargs = mock_run.call_args[1] + assert call_kwargs["force_dirty"] == {"output.txt"} + + def test_regenerate_multiple_targets(self, cli_project: Path) -> None: + build_result = BuildResult( + built=["output.txt", "image.png"], skipped=[], failed={} + ) + with ( + patch("hokusai.cli.Path") as mock_path_cls, + patch( + "hokusai.cli.run_build", + new_callable=AsyncMock, + return_value=build_result, + ) as mock_run, + ): + mock_path_cls.cwd.return_value = cli_project + result = runner.invoke(app, ["regenerate", "output.txt", "image.png"]) + + assert result.exit_code == 0 + assert "2 regenerated" in result.output + call_kwargs = mock_run.call_args[1] + assert call_kwargs["force_dirty"] == {"output.txt", "image.png"} + + def test_regenerate_unknown_target(self, cli_project: Path) -> None: + with patch("hokusai.cli.Path") as mock_path_cls: + mock_path_cls.cwd.return_value = cli_project + result = runner.invoke(app, ["regenerate", "nonexistent.txt"]) + + assert result.exit_code == 1 + assert "Unknown target(s): nonexistent.txt" in result.output + + class TestCleanCommand: """Test the clean CLI command."""