Coverage for flowr / infrastructure / config.py: 100%

76 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-05-02 18:23 +0000

1"""Configuration resolution for flowr CLI.""" 

2 

3from dataclasses import dataclass 

4from pathlib import Path 

5from typing import Any 

6 

7try: 

8 import tomllib 

9except ImportError: # pragma: no cover 

10 import tomli as tomllib # pragma: no cover 

11 

12 

13@dataclass(frozen=True, slots=True) 

14class FlowrConfig: 

15 """Resolved flowr configuration.""" 

16 

17 flows_dir: Path 

18 sessions_dir: Path 

19 default_flow: str 

20 default_session: str 

21 project_root: Path 

22 

23 def flows_path(self) -> Path: 

24 """Return the resolved flows directory as a Path.""" 

25 return self.project_root / self.flows_dir 

26 

27 def sessions_path(self) -> Path: 

28 """Return the resolved sessions directory as a Path.""" 

29 return self.project_root / self.sessions_dir 

30 

31 

32class ConfigError(Exception): 

33 """Raised when flowr configuration cannot be resolved.""" 

34 

35 

36_CONFIG_DEFAULTS: dict[str, Any] = { 

37 "flows_dir": Path(".flowr/flows"), 

38 "sessions_dir": Path(".flowr/sessions"), 

39 "default_flow": "main-flow", 

40 "default_session": "default", 

41} 

42 

43_CONFIG_KEYS = ["flows_dir", "sessions_dir", "default_flow", "default_session"] 

44 

45 

46def _read_pyproject(root: Path) -> dict[str, Any]: 

47 """Read [tool.flowr] values from pyproject.toml. 

48 

49 Returns: 

50 Dict of config values found in pyproject.toml. 

51 

52 Raises: 

53 ConfigError: If pyproject.toml exists but is malformed. 

54 """ 

55 pyproject_path = root / "pyproject.toml" 

56 if not pyproject_path.exists(): 

57 return {} 

58 

59 try: 

60 with pyproject_path.open("rb") as f: 

61 data = tomllib.load(f) 

62 except (tomllib.TOMLDecodeError, OSError) as exc: 

63 raise ConfigError(f"Failed to read {pyproject_path}: {exc}") from exc 

64 

65 tool_flowr = data.get("tool", {}).get("flowr", {}) 

66 file_values: dict[str, Any] = {} 

67 if "flows_dir" in tool_flowr: 

68 file_values["flows_dir"] = Path(tool_flowr["flows_dir"]) 

69 if "sessions_dir" in tool_flowr: 

70 file_values["sessions_dir"] = Path(tool_flowr["sessions_dir"]) 

71 if "default_flow" in tool_flowr: 

72 file_values["default_flow"] = tool_flowr["default_flow"] 

73 if "default_session" in tool_flowr: 

74 file_values["default_session"] = tool_flowr["default_session"] 

75 return file_values 

76 

77 

78def _resolve_values( 

79 file_values: dict[str, Any], 

80 overrides: dict[str, Any], 

81) -> dict[str, Any]: 

82 """Resolve config values with CLI > pyproject.toml > defaults priority.""" 

83 resolved: dict[str, Any] = {} 

84 for key in _CONFIG_KEYS: 

85 resolved[key] = overrides.get(key, file_values.get(key, _CONFIG_DEFAULTS[key])) 

86 return resolved 

87 

88 

89def _resolve_sources( 

90 file_values: dict[str, Any], 

91 overrides: dict[str, Any], 

92) -> dict[str, str]: 

93 """Determine the source of each config value.""" 

94 sources: dict[str, str] = {} 

95 for key in _CONFIG_KEYS: 

96 if key in overrides: 

97 sources[key] = "cli" 

98 elif key in file_values: 

99 sources[key] = "pyproject.toml" 

100 else: 

101 sources[key] = "default" 

102 sources["project_root"] = "cwd" 

103 return sources 

104 

105 

106def _to_config(resolved: dict[str, Any], root: Path) -> FlowrConfig: 

107 """Convert resolved values dict to a FlowrConfig instance.""" 

108 flows_dir = resolved["flows_dir"] 

109 sessions_dir = resolved["sessions_dir"] 

110 if isinstance(flows_dir, str): 

111 flows_dir = Path(flows_dir) 

112 if isinstance(sessions_dir, str): 

113 sessions_dir = Path(sessions_dir) 

114 return FlowrConfig( 

115 flows_dir=flows_dir, 

116 sessions_dir=sessions_dir, 

117 default_flow=resolved["default_flow"], 

118 default_session=resolved["default_session"], 

119 project_root=root, 

120 ) 

121 

122 

123def resolve_config( 

124 project_root: Path | None = None, 

125 cli_overrides: dict[str, Any] | None = None, 

126) -> FlowrConfig: 

127 """Resolve flowr configuration from pyproject.toml with CLI overrides. 

128 

129 Reads [tool.flowr] from pyproject.toml in project_root (or CWD if not 

130 given). CLI overrides take precedence over pyproject.toml values, which 

131 take precedence over defaults. 

132 

133 Args: 

134 project_root: Root directory containing pyproject.toml. Defaults to CWD. 

135 cli_overrides: Dict of CLI flag overrides. Supported keys: 

136 flows_dir, sessions_dir, default_flow, default_session. 

137 

138 Returns: 

139 FlowrConfig with resolved values. 

140 

141 Raises: 

142 ConfigError: If pyproject.toml exists but [tool.flowr] is malformed. 

143 """ 

144 root = project_root or Path.cwd() 

145 overrides = cli_overrides or {} 

146 file_values = _read_pyproject(root) 

147 resolved = _resolve_values(file_values, overrides) 

148 return _to_config(resolved, root) 

149 

150 

151def resolve_config_with_sources( 

152 project_root: Path | None = None, 

153 cli_overrides: dict[str, Any] | None = None, 

154) -> tuple[FlowrConfig, dict[str, str]]: 

155 """Resolve flowr configuration and track the source of each value. 

156 

157 Returns: 

158 Tuple of (FlowrConfig, sources_dict) where sources_dict maps 

159 each key to "cli", "pyproject.toml", or "default". 

160 

161 Raises: 

162 ConfigError: If pyproject.toml exists but [tool.flowr] is malformed. 

163 """ 

164 root = project_root or Path.cwd() 

165 overrides = cli_overrides or {} 

166 file_values = _read_pyproject(root) 

167 resolved = _resolve_values(file_values, overrides) 

168 sources = _resolve_sources(file_values, overrides) 

169 config = _to_config(resolved, root) 

170 return config, sources