Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/__main__.py: 91%
174 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"""Provides the main entry point and lifecycle orchestration for the Bijux CLI.
6This module is the primary entry point when the CLI is executed. It is
7responsible for orchestrating the entire lifecycle of a command invocation,
8from initial setup to final exit.
10Key responsibilities include:
11 * **Environment Setup:** Configures structured logging (`structlog`) and
12 disables terminal colors for tests.
13 * **Argument Pre-processing:** Cleans and validates command-line arguments
14 before they are passed to the command parser.
15 * **Service Initialization:** Initializes the dependency injection container,
16 registers all default services, and starts the core `Engine`.
17 * **Application Assembly:** Builds the main `Typer` application, including
18 all commands and dynamic plugins.
19 * **Execution and Error Handling:** Invokes the Typer application, catches
20 all top-level exceptions (including `Typer` errors, custom `CommandError`
21 exceptions, and `KeyboardInterrupt`), and translates them into
22 structured error messages and standardized exit codes.
23 * **History Recording:** Persists the command to the history service after
24 execution.
25"""
27from __future__ import annotations
29import contextlib
30from contextlib import suppress
31import importlib.metadata as importlib_metadata
32import io
33import json
34import logging
35import os
36import sys
37import time
38from typing import IO, Any, AnyStr
40import click
41from click.exceptions import NoSuchOption, UsageError
42import structlog
43import typer
45from bijux_cli.cli import build_app
46from bijux_cli.core.di import DIContainer
47from bijux_cli.core.engine import Engine
48from bijux_cli.core.enums import OutputFormat
49from bijux_cli.core.exceptions import CommandError
50from bijux_cli.services import register_default_services
51from bijux_cli.services.history import History
53_orig_stderr = sys.stderr
54_orig_click_echo = click.echo
55_orig_click_secho = click.secho
58class _FilteredStderr(io.TextIOBase):
59 """A proxy for `sys.stderr` that filters a specific noisy plugin warning."""
61 def write(self, data: str) -> int:
62 """Writes data to stderr, suppressing a specific known warning.
64 Args:
65 data (str): The string to write.
67 Returns:
68 int: The number of characters written, or 0 if suppressed.
69 """
70 noise = "Plugin 'test-src' does not expose a Typer app via 'cli()' or 'app'"
71 if noise in data:
72 return 0
74 if _orig_stderr.closed:
75 return 0
77 return _orig_stderr.write(data)
79 def flush(self) -> None:
80 """Flushes the underlying stderr stream."""
81 if not _orig_stderr.closed: 81 ↛ exitline 81 didn't return from function 'flush' because the condition on line 81 was always true
82 _orig_stderr.flush()
84 def __getattr__(self, name: str) -> Any:
85 """Delegates attribute access to the original `sys.stderr`.
87 Args:
88 name (str): The name of the attribute to access.
90 Returns:
91 Any: The attribute from the original `sys.stderr`.
92 """
93 return getattr(_orig_stderr, name)
96sys.stderr = _FilteredStderr()
99def _filtered_echo(
100 message: Any = None,
101 file: IO[AnyStr] | None = None,
102 nl: bool = True,
103 err: bool = False,
104 color: bool | None = None,
105 **styles: Any,
106) -> None:
107 """A replacement for `click.echo` that filters a known plugin warning.
109 Args:
110 message (Any): The message to print.
111 file (IO[AnyStr] | None): The file to write to.
112 nl (bool): If True, appends a newline.
113 err (bool): If True, writes to stderr instead of stdout.
114 color (bool | None): If True, enables color output.
115 **styles (Any): Additional style arguments for colored output.
117 Returns:
118 None
119 """
120 text = "" if message is None else str(message)
121 if (
122 text.startswith("[WARN] Plugin 'test-src'")
123 and "does not expose a Typer app" in text
124 ):
125 return
127 if styles: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 _orig_click_secho(message, file=file, nl=nl, err=err, color=color, **styles)
129 else:
130 _orig_click_echo(message, file=file, nl=nl, err=err, color=color)
133click.echo = _filtered_echo
134click.secho = _filtered_echo
135typer.echo = _filtered_echo
136typer.secho = _filtered_echo
139def disable_cli_colors_for_test() -> None:
140 """Disables color output from various libraries for test environments.
142 This function checks for the `BIJUXCLI_TEST_MODE` environment variable and,
143 if set, attempts to disable color output to ensure clean, predictable
144 test results.
145 """
146 if os.environ.get("BIJUXCLI_TEST_MODE") != "1":
147 return
148 os.environ["NO_COLOR"] = "1"
149 try:
150 from rich.console import Console
152 Console().no_color = True
153 except ImportError:
154 pass
155 try:
156 import colorama
158 colorama.deinit()
159 except ImportError:
160 pass
161 try:
162 import prompt_toolkit
164 prompt_toolkit.shortcuts.set_title = lambda text: None
165 except ImportError: # pragma: no cover
166 pass
169def should_record_command_history(command_line: list[str]) -> bool:
170 """Determines whether the given command should be recorded in the history.
172 History recording is disabled under the following conditions:
173 * The `BIJUXCLI_DISABLE_HISTORY` environment variable is set to "1".
174 * The command line is empty.
175 * The command is "history" or "help".
177 Args:
178 command_line (list[str]): The list of command-line input tokens.
180 Returns:
181 bool: True if the command should be recorded, otherwise False.
182 """
183 if os.environ.get("BIJUXCLI_DISABLE_HISTORY") == "1":
184 return False
185 if not command_line:
186 return False
187 return command_line[0].lower() not in {"history", "help"}
190def is_quiet_mode(args: list[str]) -> bool:
191 """Checks if the CLI was invoked with a quiet flag.
193 Args:
194 args (list[str]): The list of command-line arguments.
196 Returns:
197 bool: True if `--quiet` or `-q` is present, otherwise False.
198 """
199 return any(arg in ("--quiet", "-q") for arg in args)
202def print_json_error(msg: str, code: int = 2, quiet: bool = False) -> None:
203 """Prints a structured JSON error message.
205 The message is printed to stdout for usage errors (code 2) and stderr for
206 all other errors, unless quiet mode is enabled.
208 Args:
209 msg (str): The error message.
210 code (int): The error code to include in the JSON payload.
211 quiet (bool): If True, suppresses all output.
212 """
213 if not quiet:
214 print(
215 json.dumps({"error": msg, "code": code}),
216 file=sys.stdout if code == 2 else sys.stderr,
217 )
220def get_usage_for_args(args: list[str], app: typer.Typer) -> str:
221 """Gets the CLI help message for a given set of arguments.
223 This function simulates invoking the CLI with `--help` to capture the
224 contextual help message without exiting the process.
226 Args:
227 args (list[str]): The CLI arguments leading up to the help flag.
228 app (typer.Typer): The `Typer` application instance.
230 Returns:
231 str: The generated help/usage message.
232 """
233 from contextlib import redirect_stdout
234 import io
236 subcmds = []
237 for arg in args:
238 if arg in ("--help", "-h"):
239 break
240 subcmds.append(arg)
242 with io.StringIO() as buf, redirect_stdout(buf):
243 with suppress(SystemExit):
244 app(subcmds + ["--help"], standalone_mode=False)
245 return buf.getvalue()
248def _strip_format_help(args: list[str]) -> list[str]:
249 """Removes an ambiguous `--format --help` combination from arguments.
251 This prevents a parsing error where `--help` could be interpreted as the
252 value for the `--format` option.
254 Args:
255 args (list[str]): The original list of command-line arguments.
257 Returns:
258 list[str]: A filtered list of arguments.
259 """
260 new_args = []
261 skip_next = False
262 for i, arg in enumerate(args):
263 if skip_next:
264 skip_next = False
265 continue
266 if (
267 arg in ("--format", "-f")
268 and i + 1 < len(args)
269 and args[i + 1] in ("--help", "-h")
270 ):
271 skip_next = True
272 continue
273 new_args.append(arg)
274 return new_args
277def check_missing_format_argument(args: list[str]) -> str | None:
278 """Checks if a `--format` or `-f` flag is missing its required value.
280 Args:
281 args (list[str]): The list of command-line arguments.
283 Returns:
284 str | None: An error message if the value is missing, otherwise None.
285 """
286 for i, arg in enumerate(args):
287 if arg in ("--format", "-f"):
288 if i + 1 >= len(args):
289 return "Option '--format' requires an argument"
290 next_arg = args[i + 1]
291 if next_arg.startswith("-"):
292 return "Option '--format' requires an argument"
293 return None
296def setup_structlog(debug: bool = False) -> None:
297 """Configures `structlog` for the application.
299 Args:
300 debug (bool): If True, configures human-readable console output at the
301 DEBUG level. If False, configures JSON output at the CRITICAL level.
302 """
303 level = logging.DEBUG if debug else logging.CRITICAL
304 logging.basicConfig(level=level, stream=sys.stderr, format="%(message)s")
306 structlog.configure(
307 processors=[
308 structlog.contextvars.merge_contextvars,
309 structlog.stdlib.add_log_level,
310 structlog.processors.TimeStamper(fmt="iso"),
311 structlog.processors.UnicodeDecoder(),
312 (
313 structlog.dev.ConsoleRenderer()
314 if debug
315 else structlog.processors.JSONRenderer()
316 ),
317 ],
318 logger_factory=structlog.stdlib.LoggerFactory(),
319 wrapper_class=structlog.stdlib.BoundLogger,
320 cache_logger_on_first_use=True,
321 )
324def main() -> int:
325 """The main entry point for the Bijux CLI.
327 This function orchestrates the entire lifecycle of a CLI command, from
328 argument parsing and setup to execution and history recording.
330 Returns:
331 int: The final exit code of the command.
332 * `0`: Success.
333 * `1`: A generic command error occurred.
334 * `2`: A usage error or invalid option was provided.
335 * `130`: The process was interrupted by the user (Ctrl+C).
336 """
337 args = _strip_format_help(sys.argv[1:])
339 quiet = is_quiet_mode(args)
340 if quiet:
341 with contextlib.suppress(Exception):
342 sys.stderr = open(os.devnull, "w") # noqa: SIM115
343 debug = "--debug" in sys.argv or os.environ.get("BIJUXCLI_DEBUG") == "1"
344 setup_structlog(debug)
345 disable_cli_colors_for_test()
347 if any(a in ("--version", "-V") for a in args): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 try:
349 ver = importlib_metadata.version("bijux-cli")
350 except importlib_metadata.PackageNotFoundError:
351 ver = "unknown"
352 print(json.dumps({"version": ver}))
353 return 0
355 container = DIContainer.current()
356 register_default_services(
357 container, debug=False, output_format=OutputFormat.JSON, quiet=False
358 )
360 Engine()
361 app = build_app()
363 if any(a in ("-h", "--help") for a in args):
364 print(get_usage_for_args(args, app))
365 return 0
367 missing_format_msg = check_missing_format_argument(args)
368 if missing_format_msg:
369 print_json_error(missing_format_msg, 2, quiet)
370 return 2
372 command_line = args
373 start = time.time()
374 exit_code = 0
376 try:
377 result = app(args=command_line, standalone_mode=False)
378 exit_code = int(result) if isinstance(result, int) else 0
379 except typer.Exit as exc:
380 exit_code = exc.exit_code
381 except NoSuchOption as exc:
382 print_json_error(f"No such option: {exc.option_name}", 2, quiet)
383 exit_code = 2
384 except UsageError as exc:
385 print_json_error(str(exc), 2, quiet)
386 exit_code = 2
387 except CommandError as exc:
388 print_json_error(str(exc), 1, quiet)
389 exit_code = 1
390 except KeyboardInterrupt:
391 print_json_error("Aborted by user", 130, quiet)
392 exit_code = 130
393 except Exception as exc:
394 print_json_error(f"Unexpected error: {exc}", 1, quiet)
395 exit_code = 1
397 if should_record_command_history(command_line):
398 try:
399 history_service = container.resolve(History)
400 history_service.add(
401 command=" ".join(command_line),
402 params=command_line[1:],
403 success=(exit_code == 0),
404 return_code=exit_code,
405 duration_ms=int((time.time() - start) * 1000),
406 )
407 except Exception as exc:
408 print(f"[error] Could not record command history: {exc}", file=sys.stderr)
409 exit_code = 1
411 return exit_code
414if __name__ == "__main__":
415 sys.exit(main()) # pragma: no cover