Search workflow catalogs.
(
query: str | None = typer.Argument(None, help="Search query"),
tag: str | None = typer.Option(None, "--tag", help="Filter by tag"),
)
| 880 | |
| 881 | @workflow_app.command("search") |
| 882 | def workflow_search( |
| 883 | query: str | None = typer.Argument(None, help="Search query"), |
| 884 | tag: str | None = typer.Option(None, "--tag", help="Filter by tag"), |
| 885 | ): |
| 886 | """Search workflow catalogs.""" |
| 887 | from .catalog import WorkflowCatalog, WorkflowCatalogError |
| 888 | |
| 889 | project_root = _require_specify_project() |
| 890 | catalog = WorkflowCatalog(project_root) |
| 891 | |
| 892 | try: |
| 893 | results = catalog.search(query=query, tag=tag) |
| 894 | except WorkflowCatalogError as exc: |
| 895 | console.print(f"[red]Error:[/red] {exc}") |
| 896 | raise typer.Exit(1) |
| 897 | |
| 898 | if not results: |
| 899 | console.print("[yellow]No workflows found.[/yellow]") |
| 900 | return |
| 901 | |
| 902 | console.print(f"\n[bold cyan]Workflows ({len(results)}):[/bold cyan]\n") |
| 903 | for wf in results: |
| 904 | console.print(f" [bold]{wf.get('name', wf.get('id', '?'))}[/bold] ({wf.get('id', '?')}) v{wf.get('version', '?')}") |
| 905 | desc = wf.get("description", "") |
| 906 | if desc: |
| 907 | console.print(f" {desc}") |
| 908 | tags = wf.get("tags", []) |
| 909 | if tags: |
| 910 | console.print(f" [dim]Tags: {', '.join(tags)}[/dim]") |
| 911 | console.print() |
| 912 | |
| 913 | |
| 914 | @workflow_app.command("info") |
nothing calls this directly
no test coverage detected