Get information about a specific pipeline class. Args: class_name: The pipeline class name Returns: Dictionary with pipeline information including: - name: Class name - aliases: List of task aliases - supports_single_file: Whether from_single_fi
(class_name: str)
| 503 | |
| 504 | |
| 505 | def get_pipeline_info(class_name: str) -> Dict[str, Any]: |
| 506 | """ |
| 507 | Get information about a specific pipeline class. |
| 508 | |
| 509 | Args: |
| 510 | class_name: The pipeline class name |
| 511 | |
| 512 | Returns: |
| 513 | Dictionary with pipeline information including: |
| 514 | - name: Class name |
| 515 | - aliases: List of task aliases |
| 516 | - supports_single_file: Whether from_single_file() is available |
| 517 | - docstring: Class docstring (if available) |
| 518 | """ |
| 519 | registry = get_pipeline_registry() |
| 520 | aliases = get_task_aliases() |
| 521 | |
| 522 | if class_name not in registry: |
| 523 | raise ValueError(f"Unknown pipeline: {class_name}") |
| 524 | |
| 525 | cls = registry[class_name] |
| 526 | |
| 527 | # Find all aliases for this pipeline |
| 528 | pipeline_aliases = [] |
| 529 | for alias, classes in aliases.items(): |
| 530 | if class_name in classes: |
| 531 | pipeline_aliases.append(alias) |
| 532 | |
| 533 | return { |
| 534 | 'name': class_name, |
| 535 | 'aliases': pipeline_aliases, |
| 536 | 'supports_single_file': hasattr(cls, 'from_single_file'), |
| 537 | 'docstring': cls.__doc__[:200] if cls.__doc__ else None |
| 538 | } |
nothing calls this directly
no test coverage detected