Uninstall a custom step type.
(
step_id: str = typer.Argument(..., help="Step type ID to uninstall"),
)
| 1445 | |
| 1446 | @workflow_step_app.command("remove") |
| 1447 | def workflow_step_remove( |
| 1448 | step_id: str = typer.Argument(..., help="Step type ID to uninstall"), |
| 1449 | ): |
| 1450 | """Uninstall a custom step type.""" |
| 1451 | from .catalog import StepRegistry, StepValidationError |
| 1452 | |
| 1453 | project_root = _require_specify_project() |
| 1454 | |
| 1455 | _validate_step_id_or_exit(step_id) |
| 1456 | |
| 1457 | registry = StepRegistry(project_root) |
| 1458 | in_registry = registry.is_installed(step_id) |
| 1459 | |
| 1460 | steps_base_dir = _resolve_steps_base_dir_or_exit(project_root) |
| 1461 | step_dir = (steps_base_dir / step_id).resolve() |
| 1462 | # Defense-in-depth: even though _validate_step_id_or_exit rejects path |
| 1463 | # separators, ensure that the resolved directory is a single child of |
| 1464 | # steps_base_dir and is not steps_base_dir itself. |
| 1465 | try: |
| 1466 | rel_parts = step_dir.relative_to(steps_base_dir).parts |
| 1467 | except ValueError: |
| 1468 | console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") |
| 1469 | raise typer.Exit(1) |
| 1470 | if rel_parts != (step_id,): |
| 1471 | console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") |
| 1472 | raise typer.Exit(1) |
| 1473 | |
| 1474 | dir_exists = step_dir.exists() |
| 1475 | |
| 1476 | if not in_registry and not dir_exists: |
| 1477 | console.print(f"[red]Error:[/red] Step type '{step_id}' is not installed") |
| 1478 | raise typer.Exit(1) |
| 1479 | |
| 1480 | if not in_registry and dir_exists: |
| 1481 | # The registry was likely reset due to corruption. Warn the user that the |
| 1482 | # directory is being removed even though there is no registry entry, so |
| 1483 | # the orphaned package can be cleaned up and a fresh install attempted. |
| 1484 | console.print( |
| 1485 | f"[yellow]Warning:[/yellow] '{step_id}' has no registry entry " |
| 1486 | "(registry may have been reset). Removing the orphaned directory." |
| 1487 | ) |
| 1488 | |
| 1489 | if dir_exists and not in_registry: |
| 1490 | # No registry write needed; just delete the orphaned directory. |
| 1491 | import shutil |
| 1492 | try: |
| 1493 | shutil.rmtree(step_dir) |
| 1494 | except OSError as exc: |
| 1495 | console.print( |
| 1496 | f"[red]Error:[/red] Failed to remove step directory {step_dir}: {exc}" |
| 1497 | ) |
| 1498 | raise typer.Exit(1) |
| 1499 | elif in_registry: |
| 1500 | # Remove the registry entry, then the directory. If the directory |
| 1501 | # delete fails, restore the registry entry so state stays consistent |
| 1502 | # and a future `step add` isn't blocked by an orphaned directory |
| 1503 | # with no registry entry. |
| 1504 | registry_metadata = registry.get(step_id) |
no test coverage detected