Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/dev/di.py: 100%
76 statements
« prev ^ index » next coverage.py v7.10.4, created at 2025-08-19 23:36 +0000
1# SPDX-License-Identifier: MIT
2# Copyright © 2025 Bijan Mousavi
4"""Implements the `dev di` subcommand for the Bijux CLI.
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.
11Output Contract:
12 * Success: `{"factories": list, "services": list}`
13 * Verbose: Adds `{"python": str, "platform": str}` to the payload.
14 * Error: `{"error": str, "code": int}`
16Exit Codes:
17 * `0`: Success.
18 * `1`: A fatal internal error occurred (e.g., during serialization).
19 * `2`: An invalid argument or environment setting was provided (e.g.,
20 bad output path, unreadable config, invalid limit).
21 * `3`: An ASCII or encoding error was detected in the environment.
22"""
24from __future__ import annotations
26import json
27import os
28from pathlib import Path
29import platform
30from typing import Any
32import typer
33import yaml
35from bijux_cli.commands.utilities import (
36 ascii_safe,
37 emit_error_and_exit,
38 new_run_command,
39 validate_common_flags,
40)
41from bijux_cli.core.constants import (
42 HELP_DEBUG,
43 HELP_FORMAT,
44 HELP_NO_PRETTY,
45 HELP_QUIET,
46 HELP_VERBOSE,
47)
48from bijux_cli.core.di import DIContainer
50QUIET_OPTION = typer.Option(False, "-q", "--quiet", help=HELP_QUIET)
51VERBOSE_OPTION = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE)
52FORMAT_OPTION = typer.Option("json", "-f", "--format", help=HELP_FORMAT)
53PRETTY_OPTION = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY)
54DEBUG_OPTION = typer.Option(False, "-d", "--debug", help=HELP_DEBUG)
55OUTPUT_OPTION = typer.Option(
56 None,
57 "-o",
58 "--output",
59 help="Write result to file(s). May be provided multiple times.",
60)
63def _key_to_name(key: object) -> str:
64 """Converts a DI container key to its string name for serialization.
66 Args:
67 key (object): The key to convert, typically a class type or string.
69 Returns:
70 str: The string representation of the key.
71 """
72 if isinstance(key, str):
73 return key
74 name = getattr(key, "__name__", None)
75 return str(name) if name else str(key)
78def _build_dev_di_payload(include_runtime: bool) -> dict[str, Any]:
79 """Builds the DI graph payload for structured output.
81 Args:
82 include_runtime (bool): If True, includes Python and platform runtime
83 metadata in the payload.
85 Returns:
86 dict[str, Any]: A dictionary containing lists of registered 'factories'
87 and 'services', along with optional runtime information.
88 """
89 di = DIContainer.current()
91 factories = [
92 {"protocol": _key_to_name(protocol), "alias": alias}
93 for protocol, alias in di.factories()
94 ]
95 services = [
96 {"protocol": _key_to_name(protocol), "alias": alias, "implementation": None}
97 for protocol, alias in di.services()
98 ]
100 payload: dict[str, Any] = {"factories": factories, "services": services}
101 if include_runtime:
102 payload["python"] = ascii_safe(platform.python_version(), "python_version")
103 payload["platform"] = ascii_safe(platform.platform(), "platform")
104 return payload
107def dev_di_graph(
108 quiet: bool = QUIET_OPTION,
109 verbose: bool = VERBOSE_OPTION,
110 fmt: str = FORMAT_OPTION,
111 pretty: bool = PRETTY_OPTION,
112 debug: bool = DEBUG_OPTION,
113 output: list[Path] = OUTPUT_OPTION,
114) -> None:
115 """Generates and outputs the Dependency Injection (DI) container graph.
117 This developer tool inspects the DI container, validates environment
118 settings, and outputs the registration graph to stdout and/or one or more
119 files.
121 Args:
122 quiet (bool): If True, suppresses all output except for errors.
123 verbose (bool): If True, includes Python/platform details in the output.
124 fmt (str): The output format, "json" or "yaml".
125 pretty (bool): If True, pretty-prints the output.
126 debug (bool): If True, enables debug diagnostics.
127 output (list[Path]): A list of file paths to write the output to.
129 Returns:
130 None:
132 Raises:
133 SystemExit: Always exits with a contract-compliant status code and
134 payload, indicating success or detailing an error.
135 """
136 command = "dev di"
137 effective_include_runtime = (verbose or debug) and not quiet
138 effective_pretty = True if (debug and not quiet) else pretty
140 fmt_lower = validate_common_flags(
141 fmt,
142 command,
143 quiet,
144 include_runtime=effective_include_runtime,
145 )
147 limit_env = os.environ.get("BIJUXCLI_DI_LIMIT")
148 limit: int | None = None
149 if limit_env is not None:
150 try:
151 limit = int(limit_env)
152 if limit < 0:
153 emit_error_and_exit(
154 f"Invalid BIJUXCLI_DI_LIMIT value: '{limit_env}'",
155 code=2,
156 failure="limit",
157 command=command,
158 fmt=fmt_lower,
159 quiet=quiet,
160 include_runtime=effective_include_runtime,
161 debug=debug,
162 )
163 except (ValueError, TypeError):
164 emit_error_and_exit(
165 f"Invalid BIJUXCLI_DI_LIMIT value: '{limit_env}'",
166 code=2,
167 failure="limit",
168 command=command,
169 fmt=fmt_lower,
170 quiet=quiet,
171 include_runtime=effective_include_runtime,
172 debug=debug,
173 )
175 config_env = os.environ.get("BIJUXCLI_CONFIG")
176 if config_env and not config_env.isascii():
177 emit_error_and_exit(
178 f"Config path contains non-ASCII characters: {config_env!r}",
179 code=3,
180 failure="ascii",
181 command=command,
182 fmt=fmt_lower,
183 quiet=quiet,
184 include_runtime=effective_include_runtime,
185 debug=debug,
186 )
188 if config_env:
189 cfg_path = Path(config_env)
190 if cfg_path.exists() and not os.access(cfg_path, os.R_OK):
191 emit_error_and_exit(
192 f"Config path not readable: {cfg_path}",
193 code=2,
194 failure="config_unreadable",
195 command=command,
196 fmt=fmt_lower,
197 quiet=quiet,
198 include_runtime=effective_include_runtime,
199 debug=debug,
200 )
202 try:
203 payload = _build_dev_di_payload(effective_include_runtime)
204 if limit is not None:
205 payload["factories"] = payload["factories"][:limit]
206 payload["services"] = payload["services"][:limit]
207 except ValueError as exc:
208 emit_error_and_exit(
209 str(exc),
210 code=3,
211 failure="ascii",
212 command=command,
213 fmt=fmt_lower,
214 quiet=quiet,
215 include_runtime=effective_include_runtime,
216 debug=debug,
217 )
219 outputs = output
220 if outputs:
221 for p in outputs:
222 if p.is_dir():
223 emit_error_and_exit(
224 f"Output path is a directory: {p}",
225 code=2,
226 failure="output_dir",
227 command=command,
228 fmt=fmt_lower,
229 quiet=quiet,
230 include_runtime=effective_include_runtime,
231 debug=debug,
232 )
233 p.parent.mkdir(parents=True, exist_ok=True)
234 try:
235 if fmt_lower == "json":
236 p.write_text(
237 json.dumps(payload, indent=2 if effective_pretty else None)
238 + "\n",
239 encoding="utf-8",
240 )
241 else:
242 p.write_text(
243 yaml.safe_dump(
244 payload,
245 default_flow_style=False,
246 indent=2 if effective_pretty else None,
247 ),
248 encoding="utf-8",
249 )
250 except OSError as exc:
251 emit_error_and_exit(
252 f"Failed to write output file '{p}': {exc}",
253 code=2,
254 failure="output_write",
255 command=command,
256 fmt=fmt_lower,
257 quiet=quiet,
258 include_runtime=effective_include_runtime,
259 debug=debug,
260 )
262 if quiet:
263 raise typer.Exit(0)
265 if os.environ.get("BIJUXCLI_TEST_FORCE_SERIALIZE_FAIL") == "1":
266 emit_error_and_exit(
267 "Forced serialization failure",
268 code=1,
269 failure="serialize",
270 command=command,
271 fmt=fmt_lower,
272 quiet=quiet,
273 include_runtime=effective_include_runtime,
274 debug=debug,
275 )
277 new_run_command(
278 command_name=command,
279 payload_builder=lambda _: payload,
280 quiet=quiet,
281 verbose=effective_include_runtime,
282 fmt=fmt_lower,
283 pretty=effective_pretty,
284 debug=debug,
285 )