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

1# SPDX-License-Identifier: Apache-2.0 

2# Copyright © 2025 Bijan Mousavi 

3 

4"""Implements the `memory get` subcommand for the Bijux CLI. 

5 

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. 

10 

11Output Contract: 

12 * Success: `{"status": "ok", "key": str, "value": str}` 

13 * Error: `{"error": str, "code": int}` 

14 

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""" 

20 

21from __future__ import annotations 

22 

23import platform 

24 

25import typer 

26 

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 

48 

49 

50def _build_payload(include_runtime: bool, key: str, value: str) -> dict[str, object]: 

51 """Builds the payload for a single memory key-value response. 

52 

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. 

57 

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 

72 

73 

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. 

82 

83 This command validates the key's format and then fetches the corresponding 

84 value from the memory service. 

85 

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. 

92 

93 Returns: 

94 None: 

95 

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 ) 

113 

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 ) 

128 

129 memory_svc = resolve_memory_service( 

130 command, fmt_lower, quiet, include_runtime, log_level_value 

131 ) 

132 

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 ) 

159 

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 )