Coverage for  / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / commands / dev / di.py: 100%

82 statements  

« prev     ^ index     » next       coverage.py v7.13.2, created at 2026-01-26 17:59 +0000

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Implements the `dev di` subcommand for the Bijux CLI. 

5 

6This module provides a developer-focused command to introspect the internal 

7Dependency Injection (DI) container. It outputs a graph of all registered 

8service and factory protocols, which is useful for debugging the application's 

9architecture and service resolution. 

10 

11Output Contract: 

12 * Success: `{"factories": list, "services": list}` 

13 * Error: `{"error": str, "code": int}` 

14 

15Exit Codes: 

16 * `0`: Success. 

17 * `1`: A fatal internal error occurred (e.g., during serialization). 

18 * `2`: An invalid argument or environment setting was provided (e.g., 

19 bad output path, unreadable config, invalid limit). 

20 * `3`: An ASCII or encoding error was detected in the environment. 

21""" 

22 

23from __future__ import annotations 

24 

25import os 

26from pathlib import Path 

27import platform 

28from typing import Any 

29 

30import typer 

31 

32from bijux_cli.cli.core.command import ( 

33 ascii_safe, 

34 new_run_command, 

35 normalize_format, 

36 raise_exit_intent, 

37 record_history, 

38 validate_common_flags, 

39) 

40from bijux_cli.cli.core.constants import ( 

41 ENV_CONFIG, 

42 ENV_DI_LIMIT, 

43 ENV_TEST_FORCE_SERIALIZE_FAIL, 

44 OPT_FORMAT, 

45 OPT_LOG_LEVEL, 

46 OPT_PRETTY, 

47 OPT_QUIET, 

48) 

49from bijux_cli.cli.core.help_text import ( 

50 HELP_FORMAT, 

51 HELP_LOG_LEVEL, 

52 HELP_NO_PRETTY, 

53 HELP_QUIET, 

54) 

55from bijux_cli.core.di import DIContainer 

56from bijux_cli.core.enums import ErrorType, ExitCode, OutputFormat 

57from bijux_cli.core.precedence import current_execution_policy 

58 

59QUIET_OPTION = typer.Option(False, *OPT_QUIET, help=HELP_QUIET) 

60FORMAT_OPTION = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT) 

61PRETTY_OPTION = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY) 

62LOG_LEVEL_OPTION = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL) 

63OUTPUT_OPTION = typer.Option( 

64 None, 

65 "-o", 

66 "--output", 

67 help="Write result to file(s). May be provided multiple times.", 

68) 

69 

70 

71def _key_to_name(key: object) -> str: 

72 """Converts a DI container key to its string name for serialization. 

73 

74 Args: 

75 key (object): The key to convert, typically a class type or string. 

76 

77 Returns: 

78 str: The string representation of the key. 

79 """ 

80 if isinstance(key, str): 

81 return key 

82 name = getattr(key, "__name__", None) 

83 return str(name) if name else str(key) 

84 

85 

86def _build_dev_di_payload(include_runtime: bool) -> dict[str, Any]: 

87 """Builds the DI graph payload for structured output. 

88 

89 Args: 

90 include_runtime (bool): If True, includes Python and platform runtime 

91 metadata in the payload. 

92 

93 Returns: 

94 dict[str, Any]: A dictionary containing lists of registered 'factories' 

95 and 'services', along with optional runtime information. 

96 """ 

97 di = DIContainer.current() 

98 

99 factories = [ 

100 {"protocol": _key_to_name(protocol), "alias": alias} 

101 for protocol, alias in di.factories() 

102 ] 

103 services = [ 

104 {"protocol": _key_to_name(protocol), "alias": alias, "implementation": None} 

105 for protocol, alias in di.services() 

106 ] 

107 

108 payload: dict[str, Any] = {"factories": factories, "services": services} 

109 if include_runtime: 

110 payload["python"] = ascii_safe(platform.python_version(), "python_version") 

111 payload["platform"] = ascii_safe(platform.platform(), "platform") 

112 return payload 

113 

114 

115def dev_di_graph( 

116 quiet: bool = QUIET_OPTION, 

117 fmt: str = FORMAT_OPTION, 

118 pretty: bool = PRETTY_OPTION, 

119 log_level: str = LOG_LEVEL_OPTION, 

120 output: list[Path] = OUTPUT_OPTION, 

121) -> None: 

122 """Generates and outputs the Dependency Injection (DI) container graph. 

123 

124 This developer tool inspects the DI container, validates environment 

125 settings, and outputs the registration graph to stdout and/or one or more 

126 files. 

127 

128 Args: 

129 quiet (bool): If True, suppresses all output except for errors. 

130 fmt (str): The output format, "json" or "yaml". 

131 pretty (bool): If True, pretty-prints the output. 

132 log_level (str): The requested logging level. 

133 output (list[Path]): A list of file paths to write the output to. 

134 

135 Returns: 

136 None: 

137 

138 Raises: 

139 SystemExit: Always exits with a contract-compliant status code and 

140 payload, indicating success or detailing an error. 

141 """ 

142 command = "dev di" 

143 policy = current_execution_policy() 

144 quiet = policy.quiet 

145 log_level_value = policy.log_level 

146 effective_include_runtime = policy.include_runtime 

147 effective_pretty = policy.pretty 

148 fmt_lower = normalize_format(fmt) or OutputFormat.JSON 

149 

150 limit_env = os.environ.get(ENV_DI_LIMIT) 

151 limit: int | None = None 

152 if limit_env is not None: 

153 try: 

154 limit = int(limit_env) 

155 if limit < 0: 

156 raise_exit_intent( 

157 f"Invalid {ENV_DI_LIMIT} value: '{limit_env}'", 

158 code=2, 

159 failure="limit", 

160 command=command, 

161 fmt=fmt_lower, 

162 quiet=quiet, 

163 include_runtime=effective_include_runtime, 

164 log_level=log_level_value, 

165 error_type=ErrorType.USER_INPUT, 

166 ) 

167 except (ValueError, TypeError): 

168 raise_exit_intent( 

169 f"Invalid {ENV_DI_LIMIT} value: '{limit_env}'", 

170 code=2, 

171 failure="limit", 

172 command=command, 

173 fmt=fmt_lower, 

174 quiet=quiet, 

175 include_runtime=effective_include_runtime, 

176 log_level=log_level_value, 

177 error_type=ErrorType.USER_INPUT, 

178 ) 

179 

180 config_env = os.environ.get(ENV_CONFIG) 

181 if config_env and not config_env.isascii(): 

182 raise_exit_intent( 

183 f"Config path contains non-ASCII characters: {config_env!r}", 

184 code=3, 

185 failure="ascii", 

186 command=command, 

187 fmt=fmt_lower, 

188 quiet=quiet, 

189 include_runtime=effective_include_runtime, 

190 log_level=log_level_value, 

191 error_type=ErrorType.ASCII, 

192 ) 

193 

194 if config_env: 

195 cfg_path = Path(config_env) 

196 if cfg_path.exists() and not os.access(cfg_path, os.R_OK): 

197 raise_exit_intent( 

198 f"Config path not readable: {cfg_path}", 

199 code=2, 

200 failure="config_unreadable", 

201 command=command, 

202 fmt=fmt_lower, 

203 quiet=quiet, 

204 include_runtime=effective_include_runtime, 

205 log_level=log_level_value, 

206 error_type=ErrorType.USER_INPUT, 

207 ) 

208 

209 validate_common_flags( 

210 fmt_lower, 

211 command, 

212 quiet, 

213 include_runtime=effective_include_runtime, 

214 log_level=log_level_value, 

215 ) 

216 

217 try: 

218 payload = _build_dev_di_payload(effective_include_runtime) 

219 if limit is not None: 

220 payload = { 

221 "factories": payload["factories"][:limit], 

222 "services": payload["services"][:limit], 

223 **( 

224 {} 

225 if payload.get("python") is None 

226 else {"python": payload["python"]} 

227 ), 

228 **( 

229 {} 

230 if payload.get("platform") is None 

231 else {"platform": payload["platform"]} 

232 ), 

233 } 

234 except ValueError as exc: 

235 raise_exit_intent( 

236 str(exc), 

237 code=3, 

238 failure="ascii", 

239 command=command, 

240 fmt=fmt_lower, 

241 quiet=quiet, 

242 include_runtime=effective_include_runtime, 

243 log_level=log_level_value, 

244 error_type=ErrorType.ASCII, 

245 ) 

246 

247 outputs = output 

248 if outputs: 

249 for p in outputs: 

250 if p.is_dir(): 

251 raise_exit_intent( 

252 f"Output path is a directory: {p}", 

253 code=2, 

254 failure="output_dir", 

255 command=command, 

256 fmt=fmt_lower, 

257 quiet=quiet, 

258 include_runtime=effective_include_runtime, 

259 log_level=log_level_value, 

260 error_type=ErrorType.USER_INPUT, 

261 ) 

262 p.parent.mkdir(parents=True, exist_ok=True) 

263 try: 

264 from bijux_cli.cli.core.command import resolve_serializer 

265 

266 rendered = resolve_serializer().dumps( 

267 payload, fmt=fmt_lower, pretty=effective_pretty 

268 ) 

269 p.write_text(rendered.rstrip("\n") + "\n", encoding="utf-8") 

270 except OSError as exc: 

271 raise_exit_intent( 

272 f"Failed to write output file '{p}': {exc}", 

273 code=2, 

274 failure="output_write", 

275 command=command, 

276 fmt=fmt_lower, 

277 quiet=quiet, 

278 include_runtime=effective_include_runtime, 

279 log_level=log_level_value, 

280 error_type=ErrorType.USER_INPUT, 

281 ) 

282 

283 emit_output = not quiet 

284 if not emit_output: 

285 record_history(command, 0) 

286 from bijux_cli.core.exit_policy import ExitIntent, ExitIntentError 

287 

288 raise ExitIntentError( 

289 ExitIntent( 

290 code=ExitCode.SUCCESS, 

291 stream=None, 

292 payload=None, 

293 fmt=fmt_lower, 

294 pretty=effective_pretty, 

295 show_traceback=False, 

296 ) 

297 ) 

298 

299 if os.environ.get(ENV_TEST_FORCE_SERIALIZE_FAIL) == "1": 

300 raise_exit_intent( 

301 "Forced serialization failure", 

302 code=1, 

303 failure="serialize", 

304 command=command, 

305 fmt=fmt_lower, 

306 quiet=quiet, 

307 include_runtime=effective_include_runtime, 

308 log_level=log_level_value, 

309 error_type=ErrorType.INTERNAL, 

310 ) 

311 

312 new_run_command( 

313 command_name=command, 

314 payload_builder=lambda _: payload, 

315 quiet=quiet, 

316 fmt=fmt_lower, 

317 pretty=effective_pretty, 

318 log_level=log_level, 

319 )