Install a workflow from catalog, URL, or local path.
(
source: str = typer.Argument(..., help="Workflow ID, URL, or local path"),
)
| 565 | |
| 566 | @workflow_app.command("add") |
| 567 | def workflow_add( |
| 568 | source: str = typer.Argument(..., help="Workflow ID, URL, or local path"), |
| 569 | ): |
| 570 | """Install a workflow from catalog, URL, or local path.""" |
| 571 | from .catalog import WorkflowCatalog, WorkflowRegistry, WorkflowCatalogError |
| 572 | from .engine import WorkflowDefinition |
| 573 | |
| 574 | project_root = _require_specify_project() |
| 575 | registry = WorkflowRegistry(project_root) |
| 576 | workflows_dir = project_root / ".specify" / "workflows" |
| 577 | # Reject a symlinked .specify / .specify/workflows before any write so an |
| 578 | # install can't escape the project root (covers the local, URL, and |
| 579 | # catalog branches below — all write beneath workflows_dir). |
| 580 | _reject_unsafe_dir(project_root / ".specify", ".specify") |
| 581 | _reject_unsafe_dir(workflows_dir, ".specify/workflows") |
| 582 | |
| 583 | def _validate_and_install_local(yaml_path: Path, source_label: str) -> None: |
| 584 | """Validate and install a workflow from a local YAML file.""" |
| 585 | try: |
| 586 | definition = WorkflowDefinition.from_yaml(yaml_path) |
| 587 | except (ValueError, yaml.YAMLError) as exc: |
| 588 | console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}") |
| 589 | raise typer.Exit(1) |
| 590 | if not definition.id or not definition.id.strip(): |
| 591 | console.print("[red]Error:[/red] Workflow definition has an empty or missing 'id'") |
| 592 | raise typer.Exit(1) |
| 593 | |
| 594 | from .engine import validate_workflow |
| 595 | errors = validate_workflow(definition) |
| 596 | if errors: |
| 597 | console.print("[red]Error:[/red] Workflow validation failed:") |
| 598 | for err in errors: |
| 599 | console.print(f" \u2022 {err}") |
| 600 | raise typer.Exit(1) |
| 601 | |
| 602 | dest_dir = _safe_workflow_id_dir(workflows_dir, definition.id) |
| 603 | dest_dir.mkdir(parents=True, exist_ok=True) |
| 604 | import shutil |
| 605 | shutil.copy2(yaml_path, dest_dir / "workflow.yml") |
| 606 | registry.add(definition.id, { |
| 607 | "name": definition.name, |
| 608 | "version": definition.version, |
| 609 | "description": definition.description, |
| 610 | "source": source_label, |
| 611 | }) |
| 612 | console.print(f"[green]✓[/green] Workflow '{definition.name}' ({definition.id}) installed") |
| 613 | |
| 614 | # Try as URL (http/https) |
| 615 | if source.startswith("http://") or source.startswith("https://"): |
| 616 | from ipaddress import ip_address |
| 617 | from urllib.parse import urlparse |
| 618 | from specify_cli.authentication.http import open_url as _open_url |
| 619 | |
| 620 | parsed_src = urlparse(source) |
| 621 | src_host = parsed_src.hostname or "" |
| 622 | src_loopback = src_host == "localhost" |
| 623 | if not src_loopback: |
| 624 | try: |
no test coverage detected