Load a workflow from an installed ID or a local YAML path. Parameters ---------- source: Either a workflow ID (looked up in the installed workflows directory) or a path to a YAML file. Returns ------- A parsed ``WorkflowDefini
(self, source: str | Path)
| 576 | self._callback_lock = threading.Lock() |
| 577 | |
| 578 | def load_workflow(self, source: str | Path) -> WorkflowDefinition: |
| 579 | """Load a workflow from an installed ID or a local YAML path. |
| 580 | |
| 581 | Parameters |
| 582 | ---------- |
| 583 | source: |
| 584 | Either a workflow ID (looked up in the installed workflows |
| 585 | directory) or a path to a YAML file. |
| 586 | |
| 587 | Returns |
| 588 | ------- |
| 589 | A parsed ``WorkflowDefinition`` (not yet validated; call |
| 590 | ``validate_workflow()`` or ``engine.validate()`` separately). |
| 591 | |
| 592 | Raises |
| 593 | ------ |
| 594 | FileNotFoundError: |
| 595 | If the workflow file cannot be found. |
| 596 | ValueError: |
| 597 | If the workflow YAML is invalid. |
| 598 | """ |
| 599 | path = Path(source).expanduser() |
| 600 | |
| 601 | # Try as a direct file path first |
| 602 | if path.suffix.lower() in (".yml", ".yaml") and path.is_file(): |
| 603 | return WorkflowDefinition.from_yaml(path) |
| 604 | |
| 605 | # Try as an installed workflow ID |
| 606 | installed_path = ( |
| 607 | self.project_root |
| 608 | / ".specify" |
| 609 | / "workflows" |
| 610 | / str(source) |
| 611 | / "workflow.yml" |
| 612 | ) |
| 613 | if installed_path.exists(): |
| 614 | return WorkflowDefinition.from_yaml(installed_path) |
| 615 | |
| 616 | msg = f"Workflow not found: {source}" |
| 617 | raise FileNotFoundError(msg) |
| 618 | |
| 619 | def validate(self, definition: WorkflowDefinition) -> list[str]: |
| 620 | """Validate a workflow definition.""" |