Coverage for / home / runner / work / bijux-cli / bijux-cli / src / bijux_cli / cli / commands / memory / get.py: 100%
32 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-01-26 17:59 +0000
« 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 `memory get` subcommand for the Bijux CLI.
6This module contains the logic for retrieving the value of a specific key
7from the transient, in-memory data store. It provides a structured,
8machine-readable response containing the value or an error if the key is
9not found.
11Output Contract:
12 * Success: `{"status": "ok", "key": str, "value": str}`
13 * Error: `{"error": str, "code": int}`
15Exit Codes:
16 * `0`: Success.
17 * `1`: The key was not found, or another unexpected error occurred.
18 * `2`: The provided key was invalid.
19"""
21from __future__ import annotations
23import platform
25import typer
27from bijux_cli.cli.commands.memory.resolve import resolve_memory_service
28from bijux_cli.cli.core.command import (
29 ascii_safe,
30 new_run_command,
31 raise_exit_intent,
32 validate_common_flags,
33)
34from bijux_cli.cli.core.constants import (
35 OPT_FORMAT,
36 OPT_LOG_LEVEL,
37 OPT_PRETTY,
38 OPT_QUIET,
39)
40from bijux_cli.cli.core.help_text import (
41 HELP_FORMAT,
42 HELP_LOG_LEVEL,
43 HELP_NO_PRETTY,
44 HELP_QUIET,
45)
46from bijux_cli.core.enums import ErrorType
47from bijux_cli.core.precedence import current_execution_policy
50def _build_payload(include_runtime: bool, key: str, value: str) -> dict[str, object]:
51 """Builds the payload for a single memory key-value response.
53 Args:
54 include_runtime (bool): If True, includes Python and platform info.
55 key (str): The memory key that was retrieved.
56 value (str): The value associated with the key.
58 Returns:
59 Mapping[str, object]: A dictionary containing the status, key, value,
60 and optional runtime metadata.
61 """
62 payload: dict[str, object] = {"status": "ok", "key": key, "value": value}
63 if include_runtime:
64 return {
65 "status": payload["status"],
66 "key": key,
67 "value": value,
68 "python": ascii_safe(platform.python_version(), "python_version"),
69 "platform": ascii_safe(platform.platform(), "platform"),
70 }
71 return payload
74def get_memory(
75 key: str = typer.Argument(..., help="Key to retrieve"),
76 quiet: bool = typer.Option(False, *OPT_QUIET, help=HELP_QUIET),
77 fmt: str = typer.Option("json", *OPT_FORMAT, help=HELP_FORMAT),
78 pretty: bool = typer.Option(True, OPT_PRETTY, help=HELP_NO_PRETTY),
79 log_level: str = typer.Option("info", *OPT_LOG_LEVEL, help=HELP_LOG_LEVEL),
80) -> None:
81 """Retrieves a value by key from the transient in-memory store.
83 This command validates the key's format and then fetches the corresponding
84 value from the memory service.
86 Args:
87 key (str): The memory key to look up. Must be between 1 and 4096
88 printable, non-whitespace characters.
89 quiet (bool): If True, suppresses all output except for errors.
90 fmt (str): The output format, "json" or "yaml".
91 pretty (bool): If True, pretty-prints the output. log_level (str): Logging level for diagnostics.
93 Returns:
94 None:
96 Raises:
97 SystemExit: Always exits with a contract-compliant status code and
98 payload, indicating success or detailing an error.
99 """
100 command = "memory get"
101 policy = current_execution_policy()
102 quiet = policy.quiet
103 include_runtime = policy.include_runtime
104 pretty = policy.pretty
105 log_level_value = policy.log_level
106 fmt_lower = validate_common_flags(
107 fmt,
108 command,
109 quiet,
110 include_runtime=include_runtime,
111 log_level=log_level_value,
112 )
114 if not (
115 1 <= len(key) <= 4096 and all(c.isprintable() and not c.isspace() for c in key)
116 ):
117 raise_exit_intent(
118 "Invalid key: must be 1-4096 printable non-space characters",
119 code=2,
120 failure="invalid_key",
121 error_type=ErrorType.USER_INPUT,
122 command=command,
123 fmt=fmt_lower,
124 quiet=quiet,
125 include_runtime=include_runtime,
126 log_level=log_level_value,
127 )
129 memory_svc = resolve_memory_service(
130 command, fmt_lower, quiet, include_runtime, log_level_value
131 )
133 try:
134 value = memory_svc.get(key)
135 except KeyError:
136 raise_exit_intent(
137 f"Key not found: {key}",
138 code=1,
139 failure="not_found",
140 error_type=ErrorType.USER_INPUT,
141 command=command,
142 fmt=fmt_lower,
143 quiet=quiet,
144 include_runtime=include_runtime,
145 log_level=log_level_value,
146 )
147 except Exception as exc:
148 raise_exit_intent(
149 f"Failed to get memory: {exc}",
150 code=1,
151 failure="get_failed",
152 error_type=ErrorType.INTERNAL,
153 command=command,
154 fmt=fmt_lower,
155 quiet=quiet,
156 include_runtime=include_runtime,
157 log_level=log_level_value,
158 )
160 new_run_command(
161 command_name=command,
162 payload_builder=lambda include: _build_payload(include, key, value),
163 quiet=quiet,
164 fmt=fmt_lower,
165 pretty=pretty,
166 log_level=log_level,
167 )