Coverage for flowr / cli / resolution.py: 100%

1"""Flow name resolution: maps short flow names to file paths. 

2 

3If the argument is an existing file path, return it directly. 

4Otherwise, treat it as a flow name and search the configured 

5flows directory for a matching .yaml file. 

6""" 

7 

8from pathlib import Path 

9from typing import Protocol 

10 

11 

12class FlowNameNotFoundError(Exception): 

13 """Raised when a flow name cannot be resolved to a file path.""" 

14 

15 def __init__(self, flow_name: str, flows_dir: Path) -> None: 

16 """Initialize with the unresolvable flow name and searched directory.""" 

17 self.flow_name = flow_name 

18 self.flows_dir = flows_dir 

19 super().__init__(f"Flow not found: '{flow_name}' (searched in {flows_dir})") 

20 

21 

22class FlowNameResolver(Protocol): 

23 """Resolve a flow name or file path to a valid flow file path. 

24 

25 If the argument is an existing file path, return it directly. 

26 Otherwise, treat it as a flow name and search the configured 

27 flows directory for a matching .yaml file. 

28 """ 

29 

30 def resolve(self, flow_arg: str, flows_dir: Path) -> Path: # pragma: no cover 

31 """Resolve a flow argument to a file path. 

32 

33 Args: 

34 flow_arg: A file path or short flow name. 

35 flows_dir: The configured flows directory. 

36 

37 Returns: 

38 The resolved Path to the flow YAML file. 

39 

40 Raises: 

41 FlowNameNotFoundError: The argument is not an existing file 

42 and no matching .yaml file exists in flows_dir. 

43 """ 

44 ... 

45 

46 

47class DefaultFlowNameResolver: 

48 """Default implementation: file paths first, then name resolution.""" 

49 

50 def resolve(self, flow_arg: str, flows_dir: Path) -> Path: 

51 """Resolve a flow argument to a file path. 

52 

53 If flow_arg is an existing file path, return it directly 

54 (backward compatible). Otherwise, treat it as a flow name 

55 and look for {flows_dir}/{flow_name}.yaml. 

56 

57 Args: 

58 flow_arg: A file path or short flow name. 

59 flows_dir: The configured flows directory. 

60 

61 Returns: 

62 The resolved Path to the flow YAML file. 

63 

64 Raises: 

65 FlowNameNotFoundError: The argument is not an existing file 

66 and no matching .yaml file exists in flows_dir. 

67 """ 

68 path = Path(flow_arg) 

69 if path.exists(): 

70 return path 

71 

72 name = flow_arg 

73 if not name.endswith(".yaml"): 

74 name = f"{name}.yaml" 

75 

76 resolved = flows_dir / name 

77 if resolved.exists(): 

78 return resolved 

79 

80 raise FlowNameNotFoundError(flow_arg, flows_dir)