Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / core / bootstrap_flow.py: 99%
172 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
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 `UserInputError`
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 json
33import logging
34import os
35import sys
36import time
38from click.exceptions import NoSuchOption, UsageError
39import structlog
40import typer
42from bijux_cli.cli.color import set_color_mode
43from bijux_cli.cli.core.command import emit_payload, resolve_emitter, resolve_serializer
44from bijux_cli.cli.core.constants import ENV_DISABLE_HISTORY, ENV_TEST_MODE
45from bijux_cli.cli.root import build_app
46from bijux_cli.core.di import DIContainer
47from bijux_cli.core.engine import Engine
48from bijux_cli.core.enums import ErrorType, LogLevel, OutputFormat
49from bijux_cli.core.errors import UserInputError
50from bijux_cli.core.exit_policy import resolve_exit_behavior
51from bijux_cli.core.intent import CLIIntent, build_cli_intent, split_command_args
52from bijux_cli.core.precedence import (
53 EffectiveConfig,
54 ExecutionPolicy,
55 LogPolicy,
56)
57from bijux_cli.plugins.services import register_plugin_services
58from bijux_cli.services import register_default_services
59from bijux_cli.services.history import History
60from bijux_cli.services.logging.contracts import LoggingConfig
63def should_record_command_history(command_line: list[str]) -> bool:
64 """Determines whether the given command should be recorded in the history.
66 History recording is disabled under the following conditions:
67 * The `BIJUXCLI_DISABLE_HISTORY` environment variable is set to "1".
68 * The command line is empty.
69 * The command is "history" or "help".
71 Args:
72 command_line (list[str]): The list of command-line input tokens.
74 Returns:
75 bool: True if the command should be recorded, otherwise False.
76 """
77 # POLICY: history recording eligibility.
78 if os.environ.get(ENV_DISABLE_HISTORY) == "1":
79 return False
80 if not command_line:
81 return False
82 return command_line[0].lower() not in {"history", "help"}
85def get_usage_for_args(args: list[str], app: typer.Typer) -> str:
86 """Gets the CLI help message for a given set of arguments.
88 This function simulates invoking the CLI with `--help` to capture the
89 contextual help message without exiting the process.
91 Args:
92 args (list[str]): The CLI arguments leading up to the help flag.
93 app (typer.Typer): The `Typer` application instance.
95 Returns:
96 str: The generated help/usage message.
97 """
98 from contextlib import redirect_stdout
99 import io
101 subcmds = []
102 for arg in args:
103 if arg in ("--help", "-h"):
104 break
105 subcmds.append(arg)
107 # IO: capture help output by redirecting stdout.
108 with io.StringIO() as buf, redirect_stdout(buf):
109 with suppress(SystemExit):
110 app(subcmds + ["--help"], standalone_mode=False)
111 return buf.getvalue()
114def setup_structlog(log_level: LogLevel | None = None) -> None:
115 """Configures `structlog` for the application.
117 Args:
118 log_level (str | None): Optional explicit log level override.
119 """
120 # POLICY: logging level threshold for structlog.
121 level = logging.DEBUG if log_level is LogLevel.DEBUG else logging.WARNING
122 logging.basicConfig(level=level, stream=sys.stderr, format="%(message)s")
124 # IO: environment read for test-mode logging choice.
125 use_console = (log_level is LogLevel.DEBUG) or os.environ.get(ENV_TEST_MODE) == "1"
126 structlog.configure(
127 processors=[
128 structlog.contextvars.merge_contextvars,
129 structlog.stdlib.add_log_level,
130 structlog.processors.TimeStamper(fmt="iso"),
131 structlog.processors.UnicodeDecoder(),
132 structlog.dev.ConsoleRenderer()
133 if use_console
134 else structlog.processors.JSONRenderer(),
135 ],
136 logger_factory=structlog.stdlib.LoggerFactory(),
137 wrapper_class=structlog.stdlib.BoundLogger,
138 cache_logger_on_first_use=True,
139 )
142def _emit_fast_payload(
143 payload: object,
144 *,
145 fmt: OutputFormat,
146 stream: str,
147) -> None:
148 """Serialize and emit a payload without DI initialization."""
149 from dataclasses import asdict, is_dataclass
150 from typing import Any, cast
152 # FAST PATH: emit without DI initialization.
153 if is_dataclass(payload):
154 payload = asdict(cast(Any, payload))
155 if fmt is OutputFormat.YAML:
156 try:
157 import yaml
158 except ImportError:
159 text = json.dumps(payload)
160 else:
161 text = (yaml.safe_dump(payload, sort_keys=False) or "").rstrip("\n")
162 else:
163 text = json.dumps(payload)
164 # IO: direct stream output for fast path.
165 out = sys.stdout if stream == "stdout" else sys.stderr
166 print(text, file=out)
169def _emit_fast_error(
170 message: str,
171 *,
172 error_type: ErrorType,
173 quiet: bool,
174 fmt: OutputFormat,
175 log_policy: LogPolicy,
176) -> int:
177 """Emit a structured error payload without DI initialization."""
178 # POLICY: exit behavior resolved from error type and log policy.
179 behavior = resolve_exit_behavior(
180 error_type, quiet=quiet, fmt=fmt, log_policy=log_policy
181 )
182 code = int(behavior.code)
183 if behavior.stream is None:
184 return code
185 payload = {"error": message, "code": code}
186 _emit_fast_payload(payload, fmt=fmt, stream=behavior.stream)
187 return code
190def _handle_version_request(args: list[str], intent: CLIIntent) -> int | None:
191 """Handle version requests without initializing DI or plugins."""
192 if any(a in ("--version", "-V") for a in args):
193 try:
194 ver = importlib_metadata.version("bijux-cli")
195 except importlib_metadata.PackageNotFoundError:
196 ver = "unknown"
197 print(json.dumps({"version": ver}))
198 return 0
200 command, sub_args = split_command_args(args)
201 if command != "version":
202 return None
204 if "-h" in sub_args or "--help" in sub_args:
205 app = build_app(load_plugins=False)
206 print(get_usage_for_args(["version", "--help"], app))
207 return 0
209 from bijux_cli.cli.commands.version import _build_payload
211 if intent.quiet:
212 return 0
214 if intent.log_policy.show_internal:
215 print("debug: fast version path", file=sys.stderr)
217 try:
218 payload = _build_payload(intent.include_runtime)
219 except ValueError as exc:
220 return _emit_fast_error(
221 str(exc),
222 error_type=ErrorType.CONFIG,
223 quiet=intent.quiet,
224 fmt=intent.output_format,
225 log_policy=intent.log_policy,
226 )
227 _emit_fast_payload(payload, fmt=intent.output_format, stream="stdout")
228 return 0
231def _handle_help_request(args: list[str], intent: CLIIntent) -> int | None:
232 """Handle help requests without initializing DI or plugins."""
233 if not intent.help:
234 return None
235 app = build_app(load_plugins=False)
236 print(get_usage_for_args(args, app))
237 return 0
240def run_runtime(intent: CLIIntent) -> int:
241 """Run the DI/runtime execution path."""
242 # Phase: runtime init (logging config + DI graph).
243 if intent.quiet:
244 with contextlib.suppress(Exception):
245 sys.stderr = open(os.devnull, "w") # noqa: SIM115
247 logging_config = LoggingConfig(
248 quiet=intent.quiet,
249 log_level=intent.log_level,
250 color=intent.color,
251 )
253 container = DIContainer.current()
254 container.register(CLIIntent, intent)
255 container.register(EffectiveConfig, EffectiveConfig(flags=intent.flags))
256 container.register(
257 ExecutionPolicy,
258 ExecutionPolicy(
259 output_format=intent.output_format,
260 color=intent.color,
261 quiet=intent.quiet,
262 log_level=intent.log_level,
263 pretty=intent.pretty,
264 include_runtime=intent.include_runtime,
265 ),
266 )
268 register_default_services(
269 container,
270 logging_config=logging_config,
271 output_format=intent.output_format,
272 )
273 register_plugin_services(container)
275 Engine()
276 app = build_app()
277 serializer = resolve_serializer()
278 emitter = resolve_emitter()
280 # Phase: execution + emission.
281 command_line = list(intent.args)
282 start = time.time()
284 def emit_error(error_type: ErrorType, message: str) -> int:
285 behavior = resolve_exit_behavior(
286 error_type,
287 quiet=intent.quiet,
288 fmt=intent.output_format,
289 log_policy=intent.log_policy,
290 )
291 # Invariant: exit behavior is resolved once; emission just executes.
292 code = int(behavior.code)
293 if behavior.stream is None:
294 return code
295 emit_payload(
296 {"error": message, "code": code},
297 serializer=serializer,
298 emitter=emitter,
299 fmt=intent.output_format,
300 pretty=intent.pretty,
301 stream=behavior.stream,
302 )
303 return code
305 try:
306 result = app(args=command_line, standalone_mode=False)
307 exit_code = int(result) if isinstance(result, int) else 0
308 except typer.Exit as exc:
309 exit_code = exc.exit_code
310 except NoSuchOption as exc:
311 exit_code = emit_error(ErrorType.USAGE, f"No such option: {exc.option_name}")
312 except UsageError as exc:
313 exit_code = emit_error(ErrorType.USAGE, str(exc))
314 except UserInputError as exc:
315 exit_code = emit_error(ErrorType.USER_INPUT, str(exc))
316 except KeyboardInterrupt:
317 exit_code = emit_error(ErrorType.ABORTED, "Aborted by user")
318 except Exception as exc:
319 exit_code = emit_error(ErrorType.INTERNAL, f"Unexpected error: {exc}")
321 # Phase: history recording.
322 if should_record_command_history(command_line):
323 try:
324 history_service = container.resolve(History)
325 history_service.add(
326 command=" ".join(command_line),
327 params=command_line[1:],
328 success=(exit_code == 0),
329 return_code=exit_code,
330 duration_ms=int((time.time() - start) * 1000),
331 )
332 except Exception as exc:
333 print(f"[error] Could not record command history: {exc}", file=sys.stderr)
334 exit_code = 1
336 return exit_code
339def main() -> int:
340 """The main entry point for the Bijux CLI.
342 This function orchestrates the entire lifecycle of a CLI command, from
343 argument parsing and setup to execution and history recording.
345 Returns:
346 int: The final exit code of the command.
347 * `0`: Success.
348 * `1`: A generic command error occurred.
349 * `2`: A usage error or invalid option was provided.
350 * `130`: The process was interrupted by the user (Ctrl+C).
351 """
352 # Phase: intent building (no side effects).
353 args = sys.argv[1:]
354 intent = build_cli_intent(args, env=os.environ, tty=sys.stdout.isatty())
355 if intent.errors:
356 err = intent.errors[0]
357 return _emit_fast_error(
358 err.message,
359 error_type=ErrorType.USAGE,
360 quiet=intent.quiet,
361 fmt=intent.output_format,
362 log_policy=intent.log_policy,
363 )
365 # Phase: explicit fast paths.
366 fast_exit = _handle_version_request(args, intent)
367 if fast_exit is not None:
368 return fast_exit
369 fast_exit = _handle_help_request(args, intent)
370 if fast_exit is not None:
371 return fast_exit
373 # Phase: policy resolution + runtime init.
374 try:
375 DIContainer.set_log_policy(intent.log_policy)
376 setup_structlog(intent.log_level)
377 set_color_mode(intent.color)
378 except Exception as exc:
379 return _emit_fast_error(
380 f"Startup failed: {exc}",
381 error_type=ErrorType.INTERNAL,
382 quiet=intent.quiet,
383 fmt=intent.output_format,
384 log_policy=intent.log_policy,
385 )
387 # Phase: execution + emission + exit.
388 return run_runtime(intent)
391if __name__ == "__main__":
392 raise SystemExit(main()) # pragma: no cover
395__all__ = [
396 "DIContainer",
397 "Engine",
398 "main",
399 "register_default_services",
400 "register_plugin_services",
401 "run_runtime",
402 "sys",
403]