Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/commands/history/service.py: 99%
93 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 `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 * Verbose: Adds `{"python": str, "platform": str}` to the payload.
24 * Error: `{"error": str, "code": int}`
26Exit Codes:
27 * `0`: Success.
28 * `1`: A fatal error occurred (e.g., history service unavailable).
29 * `2`: An invalid argument was provided or an I/O error occurred during
30 import/export.
31"""
33from __future__ import annotations
35from collections.abc import Mapping
36import json
37from pathlib import Path
38import platform
39from typing import Any
41import typer
43from bijux_cli.commands.utilities import (
44 ascii_safe,
45 emit_error_and_exit,
46 new_run_command,
47 validate_common_flags,
48)
49from bijux_cli.contracts import HistoryProtocol
50from bijux_cli.core.constants import (
51 HELP_DEBUG,
52 HELP_FORMAT,
53 HELP_NO_PRETTY,
54 HELP_QUIET,
55 HELP_VERBOSE,
56)
57from bijux_cli.core.di import DIContainer
60def resolve_history_service(
61 command: str, fmt_lower: str, quiet: bool, include_runtime: bool, debug: bool
62) -> HistoryProtocol:
63 """Resolves the HistoryProtocol implementation from the DI container.
65 Args:
66 command (str): The full command name (e.g., "history").
67 fmt_lower (str): The chosen output format, lowercased.
68 quiet (bool): If True, suppresses non-error output.
69 include_runtime (bool): If True, includes runtime metadata in errors.
70 debug (bool): If True, enables debug diagnostics.
72 Returns:
73 HistoryProtocol: An instance of the history service.
75 Raises:
76 SystemExit: Exits with a structured error if the service cannot be
77 resolved from the container.
78 """
79 try:
80 return DIContainer.current().resolve(HistoryProtocol)
81 except Exception as exc:
82 emit_error_and_exit(
83 f"History service unavailable: {exc}",
84 code=1,
85 failure="service_unavailable",
86 command=command,
87 fmt=fmt_lower,
88 quiet=quiet,
89 include_runtime=include_runtime,
90 debug=debug,
91 )
94def history(
95 ctx: typer.Context,
96 limit: int = typer.Option(
97 20, "--limit", "-l", help="Maximum number of entries (0 means none)."
98 ),
99 group_by: str | None = typer.Option(
100 None, "--group-by", "-g", help="Group entries by a field (e.g., 'command')."
101 ),
102 filter_cmd: str | None = typer.Option(
103 None, "--filter", "-F", help="Return only entries whose command contains TEXT."
104 ),
105 sort: str | None = typer.Option(
106 None, "--sort", help="Sort key; currently only 'timestamp' is recognized."
107 ),
108 export_path: str = typer.Option(
109 None, "--export", help="Write entire history to FILE (JSON). Overwrites."
110 ),
111 import_path: str = typer.Option(
112 None, "--import", help="Load history from FILE (JSON), replacing current store."
113 ),
114 quiet: bool = typer.Option(False, "-q", "--quiet", help=HELP_QUIET),
115 verbose: bool = typer.Option(False, "-v", "--verbose", help=HELP_VERBOSE),
116 fmt: str = typer.Option("json", "-f", "--format", help=HELP_FORMAT),
117 pretty: bool = typer.Option(True, "--pretty/--no-pretty", help=HELP_NO_PRETTY),
118 debug: bool = typer.Option(False, "-d", "--debug", help=HELP_DEBUG),
119) -> None:
120 """Lists, imports, or exports the command history.
122 This function orchestrates all history-related operations. It first checks
123 for an import or export action. If neither is specified, it proceeds to
124 list the history, applying any specified filtering, grouping, or sorting.
126 Args:
127 ctx (typer.Context): The Typer context for the CLI.
128 limit (int): The maximum number of entries to return for a list operation.
129 group_by (str | None): The field to group history entries by ('command').
130 filter_cmd (str | None): A substring to filter command names by.
131 sort (str | None): The key to sort entries by ('timestamp').
132 export_path (str): The path to export history to. This is an exclusive action.
133 import_path (str): The path to import history from. This is an exclusive action.
134 quiet (bool): If True, suppresses all output except for errors.
135 verbose (bool): If True, includes Python/platform details in the output.
136 fmt (str): The output format ("json" or "yaml").
137 pretty (bool): If True, pretty-prints the output.
138 debug (bool): If True, enables debug diagnostics.
140 Returns:
141 None:
143 Raises:
144 SystemExit: Always exits with a contract-compliant status code and
145 payload upon completion or error.
146 """
147 if ctx.invoked_subcommand:
148 return
150 command = "history"
151 if debug:
152 verbose = True
153 pretty = True
154 include_runtime = verbose
156 fmt_lower = validate_common_flags(
157 fmt,
158 command,
159 quiet,
160 include_runtime=include_runtime,
161 )
163 history_svc = resolve_history_service(
164 command, fmt_lower, quiet, include_runtime, debug
165 )
167 if limit < 0:
168 emit_error_and_exit(
169 "Invalid value for --limit: must be non-negative.",
170 code=2,
171 failure="limit",
172 command=command,
173 fmt=fmt_lower,
174 quiet=quiet,
175 include_runtime=include_runtime,
176 debug=debug,
177 )
179 if sort and sort != "timestamp":
180 emit_error_and_exit(
181 "Invalid sort key: only 'timestamp' is supported.",
182 code=2,
183 failure="sort",
184 command=command,
185 fmt=fmt_lower,
186 quiet=quiet,
187 include_runtime=include_runtime,
188 debug=debug,
189 )
191 if group_by and group_by != "command":
192 emit_error_and_exit(
193 "Invalid group_by: only 'command' is supported.",
194 code=2,
195 failure="group_by",
196 command=command,
197 fmt=fmt_lower,
198 quiet=quiet,
199 include_runtime=include_runtime,
200 debug=debug,
201 )
203 if import_path:
204 try:
205 text = Path(import_path).read_text(encoding="utf-8").strip()
206 data = json.loads(text or "[]")
207 if not isinstance(data, list):
208 raise ValueError("Import file must contain a JSON array.")
209 history_svc.clear()
210 for item in data:
211 if not isinstance(item, dict):
212 continue
213 cmd = str(item.get("command") or item.get("cmd", ""))
214 cmd = ascii_safe(cmd, "command")
215 if not cmd:
216 continue
217 history_svc.add(
218 command=cmd,
219 params=item.get("params", []),
220 success=bool(item.get("success", True)),
221 return_code=item.get("return_code", 0),
222 duration_ms=item.get("duration_ms", 0.0),
223 )
224 except Exception as exc:
225 emit_error_and_exit(
226 f"Failed to import history: {exc}",
227 code=2,
228 failure="import_failed",
229 command=command,
230 fmt=fmt_lower,
231 quiet=quiet,
232 include_runtime=include_runtime,
233 debug=debug,
234 )
236 def payload_builder(_: bool) -> Mapping[str, Any]:
237 """Builds the payload confirming a successful import.
239 Args:
240 _ (bool): Unused parameter to match the expected signature.
242 Returns:
243 Mapping[str, Any]: The structured payload.
244 """
245 payload: dict[str, Any] = {"status": "imported", "file": import_path}
246 if include_runtime:
247 payload["python"] = ascii_safe(
248 platform.python_version(), "python_version"
249 )
250 payload["platform"] = ascii_safe(platform.platform(), "platform")
251 return payload
253 new_run_command(
254 command_name=command,
255 payload_builder=payload_builder,
256 quiet=quiet,
257 verbose=verbose,
258 fmt=fmt_lower,
259 pretty=pretty,
260 debug=debug,
261 )
263 if export_path:
264 try:
265 entries = history_svc.list()
266 Path(export_path).write_text(
267 json.dumps(entries, indent=2 if pretty else None) + "\n",
268 encoding="utf-8",
269 )
270 except Exception as exc:
271 emit_error_and_exit(
272 f"Failed to export history: {exc}",
273 code=2,
274 failure="export_failed",
275 command=command,
276 fmt=fmt_lower,
277 quiet=quiet,
278 include_runtime=include_runtime,
279 debug=debug,
280 )
282 def payload_builder(_: bool) -> Mapping[str, Any]:
283 """Builds the payload confirming a successful export.
285 Args:
286 _ (bool): Unused parameter to match the expected signature.
288 Returns:
289 Mapping[str, Any]: The structured payload.
290 """
291 payload: dict[str, Any] = {"status": "exported", "file": export_path}
292 if include_runtime:
293 payload["python"] = ascii_safe(
294 platform.python_version(), "python_version"
295 )
296 payload["platform"] = ascii_safe(platform.platform(), "platform")
297 return payload
299 new_run_command(
300 command_name=command,
301 payload_builder=payload_builder,
302 quiet=quiet,
303 verbose=verbose,
304 fmt=fmt_lower,
305 pretty=pretty,
306 debug=debug,
307 )
309 try:
310 entries = history_svc.list()
311 if filter_cmd:
312 entries = [e for e in entries if filter_cmd in e.get("command", "")]
313 if sort == "timestamp":
314 entries = sorted(entries, key=lambda e: e.get("timestamp", 0))
315 if group_by == "command":
316 groups: dict[str, list[dict[str, Any]]] = {}
317 for e in entries:
318 groups.setdefault(e.get("command", ""), []).append(e)
319 entries = [
320 {"group": k, "count": len(v), "entries": v} for k, v in groups.items()
321 ]
322 if limit == 0:
323 entries = []
324 elif limit > 0: 324 ↛ 339line 324 didn't jump to line 339 because the condition on line 324 was always true
325 entries = entries[-limit:]
327 except Exception as exc:
328 emit_error_and_exit(
329 f"Failed to list history: {exc}",
330 code=1,
331 failure="list_failed",
332 command=command,
333 fmt=fmt_lower,
334 quiet=quiet,
335 include_runtime=include_runtime,
336 debug=debug,
337 )
339 def list_payload_builder(include_runtime: bool) -> Mapping[str, Any]:
340 """Builds the payload containing a list of history entries.
342 Args:
343 include_runtime (bool): If True, includes Python and platform info.
345 Returns:
346 Mapping[str, Any]: The structured payload.
347 """
348 payload: dict[str, Any] = {"entries": entries}
349 if include_runtime:
350 payload["python"] = ascii_safe(platform.python_version(), "python_version")
351 payload["platform"] = ascii_safe(platform.platform(), "platform")
352 return payload
354 new_run_command(
355 command_name=command,
356 payload_builder=list_payload_builder,
357 quiet=quiet,
358 verbose=verbose,
359 fmt=fmt_lower,
360 pretty=pretty,
361 debug=debug,
362 )