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.
This commit is contained in:
Konstantin Fickel 2026-02-21 11:46:07 +01:00
parent 612ea0ae9d
commit d76951fe47
Signed by: kfickel
GPG key ID: A793722F9933C1A5
3 changed files with 124 additions and 3 deletions

View file

@ -179,6 +179,7 @@ async def run_build(
project_dir: Path, project_dir: Path,
project_name: str, project_name: str,
target: str | None = None, target: str | None = None,
force_dirty: set[str] | None = None,
on_progress: ProgressCallback = _noop_callback, on_progress: ProgressCallback = _noop_callback,
) -> BuildResult: ) -> BuildResult:
"""Execute the build. """Execute the build.
@ -186,10 +187,14 @@ async def run_build(
If *target* is specified, only build that target and its transitive If *target* is specified, only build that target and its transitive
dependencies. Otherwise build all targets. 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 Execution proceeds in topological generations each generation is a
set of independent targets that run concurrently via set of independent targets that run concurrently via
:func:`asyncio.gather`. :func:`asyncio.gather`.
""" """
force_dirty = force_dirty or set()
result = BuildResult() result = BuildResult()
providers = _create_providers() providers = _create_providers()
provider_index = _build_provider_index(providers) provider_index = _build_provider_index(providers)
@ -221,7 +226,8 @@ async def run_build(
on_progress(BuildEvent.TARGET_DEP_FAILED, name, "Dependency failed") on_progress(BuildEvent.TARGET_DEP_FAILED, name, "Dependency failed")
continue 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): if not _has_provider(name, config, provider_index, result, on_progress):
continue continue
dirty_targets.append(name) dirty_targets.append(name)

View file

@ -106,7 +106,11 @@ def _format_elapsed(seconds: float) -> str:
def _run_build( 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]: ) -> tuple[BuildResult, float]:
"""Run the async build with click-styled progress output and timing.""" """Run the async build with click-styled progress output and timing."""
@ -142,7 +146,14 @@ def _run_build(
start = time.monotonic() start = time.monotonic()
result = asyncio.run( 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 elapsed = time.monotonic() - start
@ -186,6 +197,57 @@ def build(
raise typer.Exit(code=1) 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 <project>.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() @app.command()
def clean() -> None: def clean() -> None:
"""Remove generated artifacts (targets only, not input files).""" """Remove generated artifacts (targets only, not input files)."""

View file

@ -136,6 +136,59 @@ class TestBuildCommand:
assert call_args[0][3] == "output.txt" 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: class TestCleanCommand:
"""Test the clean CLI command.""" """Test the clean CLI command."""