Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / core / precedence.py: 91%
141 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"""Flag/env/config precedence helpers."""
6from __future__ import annotations
8from collections.abc import Sequence
9from dataclasses import dataclass, field
10import sys
11import time
12from typing import Any
14from bijux_cli.core.enums import ColorMode, ErrorType, ExitCode, LogLevel, OutputFormat
15from bijux_cli.core.exit_policy import ExitIntent, resolve_exit_behavior
18@dataclass(frozen=True)
19class GlobalCLIConfig:
20 """Immutable container for parsed global CLI flags."""
22 help: bool
23 flags: FlagLayer
24 args: tuple[str, ...]
25 errors: tuple[FlagError, ...]
28@dataclass(frozen=True)
29class FlagError:
30 """Structured error for flag parsing/validation."""
32 message: str
33 failure: str
34 flag: str
37@dataclass(frozen=True)
38class Flags:
39 """Resolved flag bundle for logging/output behavior."""
41 quiet: bool
42 log_level: LogLevel
43 color: ColorMode
44 format: OutputFormat
47@dataclass(frozen=True)
48class FlagLayer:
49 """Optional flag layer for precedence resolution."""
51 log_level: LogLevel | None = None
52 color: ColorMode | None = None
53 format: OutputFormat | None = None
54 quiet: bool | None = None
57@dataclass(frozen=True)
58class EffectiveConfig:
59 """Resolved output/logging flags after precedence and normalization."""
61 flags: Flags
64@dataclass(frozen=True)
65class ExecutionPolicy:
66 """Resolved execution policy shared across CLI/service boundaries."""
68 output_format: OutputFormat
69 color: ColorMode
70 quiet: bool
71 log_level: LogLevel
72 log_policy: LogPolicy = field(init=False)
73 pretty: bool = True
74 include_runtime: bool = False
76 def __post_init__(self) -> None:
77 """Backfill log policy when constructed directly."""
78 object.__setattr__(self, "log_policy", resolve_log_policy(self.log_level))
81@dataclass(frozen=True)
82class OutputConfig:
83 """Resolved output/logging configuration for services."""
85 include_runtime: bool
86 pretty: bool
87 log_level: LogLevel
88 color: ColorMode
89 format: OutputFormat
90 log_policy: LogPolicy
93@dataclass(frozen=True)
94class LogPolicy:
95 """Typed logging policy derived from a log level threshold."""
97 level: LogLevel
98 show_internal: bool
99 show_traceback: bool
100 pretty_default: bool
101 telemetry_verbosity: int
104_LOG_RANK: dict[LogLevel, int] = {
105 LogLevel.TRACE: 5,
106 LogLevel.DEBUG: 10,
107 LogLevel.INFO: 20,
108 LogLevel.WARNING: 30,
109 LogLevel.ERROR: 40,
110 LogLevel.CRITICAL: 50,
111}
114def _log_rank(level: LogLevel) -> int:
115 """Return a comparable rank for log levels."""
116 return _LOG_RANK.get(level, _LOG_RANK[LogLevel.INFO])
119def resolve_log_policy(log_level: LogLevel) -> LogPolicy:
120 """Derive logging policy from a level threshold."""
121 rank = _log_rank(log_level)
122 debug_rank = _log_rank(LogLevel.DEBUG)
123 info_rank = _log_rank(LogLevel.INFO)
124 warn_rank = _log_rank(LogLevel.WARNING)
125 if rank <= debug_rank:
126 telemetry = 3
127 elif rank <= info_rank:
128 telemetry = 2
129 elif rank <= warn_rank:
130 telemetry = 1
131 else:
132 telemetry = 0
133 return LogPolicy(
134 level=log_level,
135 show_internal=rank <= debug_rank,
136 show_traceback=rank <= debug_rank,
137 pretty_default=rank <= info_rank,
138 telemetry_verbosity=telemetry,
139 )
142def resolve_exit_intent(
143 *,
144 message: str,
145 code: int,
146 failure: str,
147 command: str | None,
148 fmt: OutputFormat,
149 quiet: bool,
150 include_runtime: bool,
151 error_type: ErrorType,
152 log_level: LogLevel = LogLevel.INFO,
153 log_policy: LogPolicy | None = None,
154 extra: dict[str, object] | None = None,
155) -> ExitIntent:
156 """Resolve an exit intent and build a structured error payload."""
157 policy = log_policy or resolve_log_policy(log_level)
158 behavior = resolve_exit_behavior(
159 error_type,
160 quiet=quiet,
161 fmt=fmt,
162 log_policy=policy,
163 )
164 payload: dict[str, object] = {"error": message, "code": int(code)}
165 if failure: 165 ↛ 167line 165 didn't jump to line 167 because the condition on line 165 was always true
166 payload["failure"] = failure
167 if command: 167 ↛ 169line 167 didn't jump to line 169 because the condition on line 167 was always true
168 payload["command"] = command
169 if fmt: 169 ↛ 171line 169 didn't jump to line 171 because the condition on line 169 was always true
170 payload["fmt"] = fmt
171 if extra:
172 payload.update(extra)
173 if behavior.show_traceback: 173 ↛ 174line 173 didn't jump to line 174 because the condition on line 173 was never true
174 import traceback
176 trace = traceback.format_exc()
177 if "NoneType: None" not in trace:
178 payload["traceback"] = trace
179 if include_runtime:
180 payload["python"] = sys.version.split()[0]
181 payload["platform"] = sys.platform
182 payload["timestamp"] = str(time.time())
183 return ExitIntent(
184 code=ExitCode(int(code)),
185 stream=behavior.stream,
186 payload=payload,
187 fmt=fmt,
188 pretty=False,
189 show_traceback=behavior.show_traceback,
190 )
193def validate_cli_flags(
194 config: GlobalCLIConfig, parse_errors: Sequence[FlagError] | None = None
195) -> tuple[FlagError, ...]:
196 """Validate raw CLI flags without applying behavior."""
197 errors: list[FlagError] = list(parse_errors or config.errors)
198 flags = config.flags
199 if flags.format is not None and flags.format not in (
200 OutputFormat.JSON,
201 OutputFormat.YAML,
202 ):
203 errors.append(
204 FlagError(
205 message="Invalid output format.",
206 failure="invalid_format",
207 flag="--format",
208 )
209 )
210 if flags.color is not None and flags.color not in ( 210 ↛ 215line 210 didn't jump to line 215 because the condition on line 210 was never true
211 ColorMode.AUTO,
212 ColorMode.ALWAYS,
213 ColorMode.NEVER,
214 ):
215 errors.append(
216 FlagError(
217 message="Invalid color mode.",
218 failure="invalid_color",
219 flag="--color",
220 )
221 )
222 if flags.log_level is not None and flags.log_level not in ( 222 ↛ 227line 222 didn't jump to line 227 because the condition on line 222 was never true
223 LogLevel.TRACE,
224 LogLevel.DEBUG,
225 LogLevel.INFO,
226 ):
227 errors.append(
228 FlagError(
229 message="Invalid log level.",
230 failure="invalid_log_level",
231 flag="--log-level",
232 )
233 )
234 return tuple(errors)
237def _pick_value(
238 cli: FlagLayer,
239 env: FlagLayer,
240 file: FlagLayer,
241 defaults: Flags,
242) -> Flags:
243 """Resolve precedence across four layers with first-set wins."""
245 def pick(attr: str, fallback: Any) -> Any:
246 for source in (cli, env, file):
247 value = getattr(source, attr)
248 if value is not None:
249 return value
250 return fallback
252 return Flags(
253 quiet=bool(pick("quiet", defaults.quiet)),
254 log_level=pick("log_level", defaults.log_level),
255 color=pick("color", defaults.color),
256 format=pick("format", defaults.format),
257 )
260def resolve_effective_config(
261 cli: FlagLayer,
262 env: FlagLayer,
263 file: FlagLayer,
264 defaults: Flags,
265) -> EffectiveConfig:
266 """Resolve flag/env/config precedence into a single effective config.
268 Algebraic laws:
269 - Left-identity: resolve(cli, env, file, defaults) equals resolve(cli, empty, empty, defaults)
270 - Right-identity: resolve(empty, empty, empty, defaults) equals defaults
271 - Idempotence: resolve(a, a, a, defaults) equals resolve(a, empty, empty, defaults)
272 """
273 flags = _pick_value(cli, env, file, defaults)
274 if flags.quiet:
275 flags = Flags(
276 quiet=True,
277 log_level=LogLevel.ERROR,
278 color=flags.color,
279 format=flags.format,
280 )
281 return EffectiveConfig(flags=flags)
284def default_execution_policy() -> ExecutionPolicy:
285 """Return the default execution policy without DI."""
286 defaults = Flags(
287 quiet=False,
288 log_level=LogLevel.INFO,
289 color=ColorMode.AUTO,
290 format=OutputFormat.JSON,
291 )
292 effective = resolve_effective_config(
293 cli=FlagLayer(),
294 env=FlagLayer(),
295 file=FlagLayer(),
296 defaults=defaults,
297 )
298 log_policy = resolve_log_policy(effective.flags.log_level)
299 return ExecutionPolicy(
300 output_format=effective.flags.format,
301 color=effective.flags.color,
302 quiet=effective.flags.quiet,
303 log_level=effective.flags.log_level,
304 pretty=log_policy.pretty_default,
305 include_runtime=log_policy.show_internal,
306 )
309def resolve_output_flags(
310 *,
311 quiet: bool,
312 pretty: bool,
313 log_level: LogLevel = LogLevel.INFO,
314 color: ColorMode = ColorMode.AUTO,
315 output_format: OutputFormat = OutputFormat.JSON,
316) -> OutputConfig:
317 """Resolve logging/color/pretty flags from a single source of truth."""
318 effective = resolve_effective_config(
319 cli=FlagLayer(
320 quiet=quiet,
321 log_level=log_level,
322 color=color,
323 format=output_format,
324 ),
325 env=FlagLayer(),
326 file=FlagLayer(),
327 defaults=Flags(
328 quiet=False,
329 log_level=LogLevel.INFO,
330 color=ColorMode.AUTO,
331 format=OutputFormat.JSON,
332 ),
333 )
334 log_policy = resolve_log_policy(effective.flags.log_level)
335 return OutputConfig(
336 include_runtime=False,
337 pretty=pretty,
338 log_level=effective.flags.log_level,
339 color=effective.flags.color,
340 format=effective.flags.format,
341 log_policy=log_policy,
342 )
345def current_execution_policy() -> ExecutionPolicy:
346 """Resolve the execution policy from CLI intent or DI."""
347 from bijux_cli.core.di import DIContainer
348 from bijux_cli.core.intent import current_cli_intent
350 try:
351 policy_obj: object = DIContainer.current().resolve(ExecutionPolicy)
352 except Exception:
353 policy_obj = None
354 if isinstance(policy_obj, ExecutionPolicy): 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 return policy_obj
357 intent = current_cli_intent()
358 return ExecutionPolicy(
359 output_format=intent.output_format,
360 color=intent.color,
361 quiet=intent.quiet,
362 log_level=intent.log_level,
363 pretty=intent.pretty,
364 include_runtime=intent.include_runtime,
365 )