Coverage for collectra/main.py: 81.82%

1"""Collectra command-line entry point.""" 

2 

3from __future__ import annotations 

4 

5import sys 

6from pathlib import Path 

7from typing import TYPE_CHECKING 

8 

9from .cli import invoke 

10 

11if TYPE_CHECKING: 

12 from .pipelines.base import Collectra 

13 

14 

15def find_pipeline(argv: list[str]) -> tuple[Path, list[str]]: 

16 """Extract ``--pipeline PATH`` while preserving all other arguments.""" 

17 remaining = list(argv) 

18 for index, argument in enumerate(remaining): 

19 if argument == "--pipeline": 

20 if index + 1 >= len(remaining): 

21 raise SystemExit("--pipeline requires a path") 

22 path = Path(remaining[index + 1]) 

23 del remaining[index : index + 2] 

24 return path, remaining 

25 if argument.startswith("--pipeline="): 

26 value = argument.partition("=")[2] 

27 if not value: 

28 raise SystemExit("--pipeline requires a path") 

29 del remaining[index] 

30 return Path(value), remaining 

31 raise SystemExit("Missing required option: --pipeline PATH") 

32 

33 

34def resolve_workflow_path(workflow: Path) -> "Collectra": 

35 """Backward-compatible name for loading a pipeline.""" 

36 from .pipelines.base import Collectra 

37 

38 return Collectra.from_file(workflow) 

39 

40 

41def main( 

42 argv: list[str] | None = None, 

43 *, 

44 pipeline_path: Path | None = None, 

45 program_name: str = "collectra", 

46): 

47 """Load the selected pipeline and delegate its runtime object to Cappa.""" 

48 arguments = list(sys.argv[1:] if argv is None else argv) 

49 if pipeline_path is None: 

50 pipeline_path, arguments = find_pipeline(arguments) 

51 from .pipelines.base import Collectra 

52 

53 pipeline = Collectra.from_file(pipeline_path) 

54 return invoke(pipeline, arguments, name=program_name) 

55 

56 

57if __name__ == "__main__": 

58 main()