Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / commands / history / service.py: 99%
123 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"""Implements the `history` command for the Bijux CLI.
6This module provides functionality to interact with the persistent command
7history. It allows for listing, filtering, sorting, grouping, importing, and
8exporting history entries. All operations produce structured, machine-readable
9output.
11The command has three primary modes of operation:
121. **Listing (Default):** When no import/export flags are used, it lists
13 history entries, which can be filtered, sorted, and grouped.
142. **Import:** The `--import` flag replaces the current history with data
15 from a specified JSON file.
163. **Export:** The `--export` flag writes the entire current history to a
17 specified JSON file.
19Output Contract:
20 * List Success: `{"entries": list}`
21 * Import Success: `{"status": "imported", "file": str}`
22 * Export Success: `{"status": "exported", "file": str}`
23 * Error: `{"error": str, "code": int}`
25Exit Codes:
26 * `0`: Success.
27 * `1`: A fatal error occurred (e.g., history service unavailable).
28 * `2`: An invalid argument was provided or an I/O error occurred during
29 import/export.
30"""
32from __future__ import annotations
34from dataclasses import dataclass
35import json
36from pathlib import Path
37import platform
38from typing import Any
40import typer
42from bijux_cli.cli.core.command import (
43 ascii_safe,
44 new_run_command,
45 raise_exit_intent,
46 validate_common_flags,
47)
48from bijux_cli.cli.core.constants import (
49 OPT_FORMAT,
50 OPT_LOG_LEVEL,
51 OPT_PRETTY,
52 OPT_QUIET,
53)
54from bijux_cli.cli.core.help_text import (
55 HELP_FORMAT,
56 HELP_LOG_LEVEL,
57 HELP_NO_PRETTY,
58 HELP_QUIET,
59)
60from bijux_cli.core.di import DIContainer
61from bijux_cli.core.enums import ErrorType, LogLevel, OutputFormat
62from bijux_cli.core.precedence import current_execution_policy
63from bijux_cli.services.history.contracts import HistoryProtocol
66def resolve_history_service(
67 command: str,
68 fmt_lower: OutputFormat,
69 quiet: bool,
70 include_runtime: bool,
71 log_level: LogLevel,
72) -> HistoryProtocol:
73 """Resolves the HistoryProtocol implementation from the DI container.
75 Args:
76 command (str): The full command name (e.g., "history").
77 fmt_lower (OutputFormat): The chosen output format.
78 quiet (bool): If True, suppresses non-error output.
79 include_runtime (bool): If True, includes runtime metadata in errors.
80 log_level (LogLevel): Logging level for diagnostics.
82 Returns:
83 HistoryProtocol: An instance of the history service.
85 Raises:
86 SystemExit: Exits with a structured error if the service cannot be
87 resolved from the container.
88 """
89 try:
90 return DIContainer.current().resolve(HistoryProtocol)
91 except Exception as exc:
92 raise_exit_intent(
93 f"History service unavailable: {exc}",
94 code=1,
95 failure="service_unavailable",
96 error_type=ErrorType.INTERNAL,
97 command=command,
98 fmt=fmt_lower,
99 quiet=quiet,
100 include_runtime=include_runtime,
101 log_level=log_level,
102 )
105@dataclass(frozen=True)
106class HistoryIntent:
107 """Resolved intent for the history command."""
109 command: str
110 action: str
111 limit: int
112 group_by: str | None
113 filter_cmd: str | None
114 sort: str | None
115 export_path: str | None
116 import_path: str | None
117 quiet: bool
118 include_runtime: bool
119 log_level: LogLevel
120 fmt: OutputFormat
123def _build_history_intent(
124 *,
125 command: str,
126 limit: int,
127 group_by: str | None,
128 filter_cmd: str | None,
129 sort: str | None,
130 export_path: str | None,
131 import_path: str | None,
132 fmt_lower: OutputFormat,
133 quiet: bool,
134 include_runtime: bool,
135 log_level: LogLevel,
136) -> HistoryIntent:
137 """Validate inputs and build a history intent."""
138 action = "list"
139 if import_path:
140 action = "import"
141 elif export_path:
142 action = "export"
144 if limit < 0:
145 raise_exit_intent(
146 "Invalid value for --limit: must be non-negative.",
147 code=2,
148 failure="limit",
149 command=command,
150 fmt=fmt_lower,
151 quiet=quiet,
152 include_runtime=include_runtime,
153 log_level=log_level,
154 error_type=ErrorType.USER_INPUT,
155 )
157 if sort and sort != "timestamp":
158 raise_exit_intent(
159 "Invalid sort key: only 'timestamp' is supported.",
160 code=2,
161 failure="sort",
162 command=command,
163 fmt=fmt_lower,
164 quiet=quiet,
165 include_runtime=include_runtime,
166 log_level=log_level,
167 error_type=ErrorType.USER_INPUT,
168 )
170 if group_by and group_by != "command":
171 raise_exit_intent(
172 "Invalid group_by: only 'command' is supported.",
173 code=2,
174 failure="group_by",
175 command=command,
176 fmt=fmt_lower,
177 quiet=quiet,
178 include_runtime=include_runtime,
179 log_level=log_level,
180 error_type=ErrorType.USER_INPUT,
181 )
183 return HistoryIntent(
184 command=command,
185 action=action,
186 limit=limit,
187 group_by=group_by,
188 filter_cmd=filter_cmd,
189 sort=sort,
190 export_path=export_path,
191 import_path=import_path,
192 quiet=quiet,
193 include_runtime=include_runtime,
194 log_level=log_level,
195 fmt=fmt_lower,
196 )
199def _import_history(
200 intent: HistoryIntent, history_svc: HistoryProtocol
201) -> dict[str, object]:
202 """Import history data and return a payload."""
203 try:
204 text = Path(intent.import_path or "").read_text(encoding="utf-8").strip()
205 data = json.loads(text or "[]")
206 if not isinstance(data, list):
207 raise ValueError("Import file must contain a JSON array.")
208 history_svc.clear()
209 for item in data:
210 if not isinstance(item, dict):
211 continue
212 cmd = str(item.get("command") or item.get("cmd", ""))
213 cmd = ascii_safe(cmd, "command")
214 if not cmd:
215 continue
216 history_svc.add(
217 command=cmd,
218 params=item.get("params", []),
219 success=bool(item.get("success", True)),
220 return_code=item.get("return_code", 0),
221 duration_ms=item.get("duration_ms", 0.0),
222 )
223 except Exception as exc:
224 raise_exit_intent(
225 f"Failed to import history: {exc}",
226 code=2,
227 failure="import_failed",
228 command=intent.command,
229 fmt=intent.fmt,
230 quiet=intent.quiet,
231 include_runtime=intent.include_runtime,
232 log_level=intent.log_level,
233 error_type=ErrorType.USER_INPUT,
234 )
236 payload: dict[str, object] = {
237 "status": "imported",
238 "file": intent.import_path or "",
239 }
240 if intent.include_runtime:
241 return {
242 "status": payload["status"],
243 "file": payload["file"],
244 "python": ascii_safe(platform.python_version(), "python_version"),
245 "platform": ascii_safe(platform.platform(), "platform"),
246 }
247 return payload
250def _export_history(
251 intent: HistoryIntent, history_svc: HistoryProtocol
252) -> dict[str, object]:
253 """Export history data and return a payload."""
254 try:
255 entries = history_svc.list()
256 from bijux_cli.cli.core.command import resolve_serializer
258 rendered = resolve_serializer().dumps(entries, fmt=intent.fmt, pretty=True)
259 Path(intent.export_path or "").write_text(
260 rendered.rstrip("\n") + "\n",
261 encoding="utf-8",
262 )
263 except Exception as exc:
264 raise_exit_intent(
265 f"Failed to export history: {exc}",
266 code=2,
267 failure="export_failed",
268 command=intent.command,
269 fmt=intent.fmt,
270 quiet=intent.quiet,
271 include_runtime=intent.include_runtime,
272 log_level=intent.log_level,
273 error_type=ErrorType.USER_INPUT,
274 )
276 payload: dict[str, object] = {
277 "status": "exported",
278 "file": intent.export_path or "",
279 }
280 if intent.include_runtime:
281 return {
282 "status": payload["status"],
283 "file": payload["file"],
284 "python": ascii_safe(platform.python_version(), "python_version"),
285 "platform": ascii_safe(platform.platform(), "platform"),
286 }
287 return payload
290def _list_history(
291 intent: HistoryIntent, history_svc: HistoryProtocol
292) -> dict[str, object]:
293 """List history entries and return a payload."""
294 try:
295 entries = history_svc.list()
296 if intent.filter_cmd:
297 entries = [e for e in entries if intent.filter_cmd in e.get("command", "")]
298 if intent.sort == "timestamp":
299 entries = sorted(entries, key=lambda e: e.get("timestamp", 0))
300 if intent.group_by == "command":
301 groups: dict[str, list[dict[str, Any]]] = {}
302 for e in entries:
303 groups.setdefault(e.get("command", ""), []).append(e)
304 entries = [
305 {"group": k, "count": len(v), "entries": v} for k, v in groups.items()
306 ]
307 if intent.limit == 0:
308 entries = []
309 elif intent.limit > 0: 309 ↛ 323line 309 didn't jump to line 323 because the condition on line 309 was always true
310 entries = entries[-intent.limit :]
311 except Exception as exc:
312 raise_exit_intent(
313 f"Failed to list history: {exc}",
314 code=1,
315 failure="list_failed",
316 command=intent.command,
317 fmt=intent.fmt,
318 quiet=intent.quiet,
319 include_runtime=intent.include_runtime,
320 log_level=intent.log_level,
321 )
323 payload: dict[str, object] = {"entries": entries}
324 if intent.include_runtime:
325 return {
326 "entries": payload["entries"],
327 "python": ascii_safe(platform.python_version(), "python_version"),
328 "platform": ascii_safe(platform.platform(), "platform"),
329 }
330 return payload
333def _with_runtime(
334 payload: dict[str, object], include_runtime: bool
335) -> dict[str, object]:
336 """Attach runtime metadata when requested."""
337 if not include_runtime:
338 return payload
339 return {
340 **payload,
341 "python": ascii_safe(platform.python_version(), "python_version"),
342 "platform": ascii_safe(platform.platform(), "platform"),
343 }
346def history(
347 ctx: typer.Context,
348 limit: int = typer.Option(
349 20, "--limit", "-l", help="Maximum number of entries (0 means none)."
350 ),
351 group_by: str | None = typer.Option(
352 None, "--group-by", "-g", help="Group entries by a field (e.g., 'command')."
353 ),
354 filter_cmd: str | None = typer.Option(
355 None, "--filter", "-F", help="Return only entries whose command contains TEXT."
356 ),
357 sort: str | None = typer.Option(
358 None, "--sort", help="Sort key; currently only 'timestamp' is recognized."
359 ),
360 export_path: str = typer.Option(
361 None, "--export", help="Write entire history to FILE (JSON). Overwrites."
362 ),
363 import_path: str = typer.Option(
364 None, "--import", help="Load history from FILE (JSON), replacing current store."
365 ),
366 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
367 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
368 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
369 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
370) -> None:
371 """Lists, imports, or exports the command history.
373 This function orchestrates all history-related operations. It first checks
374 for an import or export action. If neither is specified, it proceeds to
375 list the history, applying any specified filtering, grouping, or sorting.
377 Args:
378 ctx (typer.Context): The Typer context for the CLI.
379 limit (int): The maximum number of entries to return for a list operation.
380 group_by (str | None): The field to group history entries by ('command').
381 filter_cmd (str | None): A substring to filter command names by.
382 sort (str | None): The key to sort entries by ('timestamp').
383 export_path (str): The path to export history to. This is an exclusive action.
384 import_path (str): The path to import history from. This is an exclusive action.
385 quiet (bool): If True, suppresses all output except for errors.
386 fmt (str): The output format ("json" or "yaml").
387 pretty (bool): If True, pretty-prints the output.
388 log_level (str): Logging level for diagnostics.
390 Returns:
391 None:
393 Raises:
394 SystemExit: Always exits with a contract-compliant status code and
395 payload upon completion or error.
396 """
397 if ctx.invoked_subcommand:
398 return
400 command = "history"
401 policy = current_execution_policy()
402 quiet = policy.quiet
403 include_runtime = policy.include_runtime
404 log_level_value = policy.log_level
405 pretty = policy.pretty
406 fmt_lower = validate_common_flags(
407 fmt,
408 command,
409 quiet,
410 include_runtime=include_runtime,
411 log_level=log_level_value,
412 )
414 history_svc = resolve_history_service(
415 command, fmt_lower, quiet, include_runtime, log_level_value
416 )
418 intent = _build_history_intent(
419 command=command,
420 limit=limit,
421 group_by=group_by,
422 filter_cmd=filter_cmd,
423 sort=sort,
424 export_path=export_path,
425 import_path=import_path,
426 fmt_lower=fmt_lower,
427 quiet=quiet,
428 include_runtime=include_runtime,
429 log_level=log_level_value,
430 )
432 payload: dict[str, object]
433 if intent.action == "import":
434 payload = _import_history(intent, history_svc)
435 elif intent.action == "export":
436 payload = _export_history(intent, history_svc)
437 else:
438 payload = _list_history(intent, history_svc)
440 new_run_command(
441 command_name=command,
442 payload_builder=lambda include_runtime: _with_runtime(payload, include_runtime),
443 quiet=quiet,
444 fmt=fmt_lower,
445 pretty=pretty,
446 log_level=log_level_value,
447 )