Coverage for /home/runner/work/bijux-cli/bijux-cli/src/bijux_cli/services/plugins/groups.py: 100%

40 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 

3 

4"""Provides helpers for defining plugin command groups and autocompletions. 

5 

6This module offers a convenient, decorator-based API for plugin developers 

7to register command groups and their subcommands. It also includes a factory 

8function for creating dynamic shell completers for command arguments, 

9enhancing the interactive user experience of plugins. 

10""" 

11 

12from __future__ import annotations 

13 

14from collections.abc import Callable 

15from typing import Any 

16 

17import typer 

18 

19from bijux_cli.contracts import ( 

20 ObservabilityProtocol, 

21 RegistryProtocol, 

22 TelemetryProtocol, 

23) 

24from bijux_cli.core.di import DIContainer 

25 

26 

27def command_group( 

28 name: str, 

29 *, 

30 version: str | None = None, 

31) -> Callable[[str], Callable[[Callable[..., Any]], Callable[..., Any]]]: 

32 """A decorator factory for registering plugin subcommands under a group. 

33 

34 This function is designed to be used as a nested decorator to easily 

35 define command groups within a plugin. 

36 

37 Example: 

38 A plugin can define a "user" command group with a "create" subcommand 

39 like this:: 

40 

41 group = command_group("user", version="1.0") 

42 

43 @group(sub="create") 

44 def create_user(username: str): 

45 ... 

46 

47 Args: 

48 name (str): The name of the parent command group (e.g., "user"). 

49 version (str | None): An optional version string for the group. 

50 

51 Returns: 

52 Callable[[str], Callable[[Callable[..., Any]], Callable[..., Any]]]: 

53 A decorator that takes a subcommand name and returns the final 

54 decorator for the function. 

55 """ 

56 

57 def with_sub(sub: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: 

58 """Captures the subcommand name for registration. 

59 

60 Args: 

61 sub (str): The name of the subcommand (e.g., "create"). 

62 

63 Returns: 

64 A decorator for the subcommand function. 

65 

66 Raises: 

67 ValueError: If the subcommand name contains spaces. 

68 """ 

69 if " " in sub: 

70 raise ValueError("subcommand may not contain spaces") 

71 full = f"{name} {sub}" 

72 

73 def decorator(fn: Callable[..., Any]) -> Callable[..., Any]: 

74 """Registers the decorated function as a command. 

75 

76 Args: 

77 fn (Callable[..., Any]): The function to register as a subcommand. 

78 

79 Returns: 

80 The original, undecorated function. 

81 

82 Raises: 

83 RuntimeError: If the `RegistryProtocol` is not initialized. 

84 """ 

85 try: 

86 di = DIContainer.current() 

87 reg: RegistryProtocol = di.resolve(RegistryProtocol) 

88 except KeyError as exc: 

89 raise RuntimeError("RegistryProtocol is not initialized") from exc 

90 

91 reg.register(full, fn, version=version) 

92 

93 try: 

94 obs: ObservabilityProtocol = di.resolve(ObservabilityProtocol) 

95 obs.log( 

96 "info", 

97 "Registered command group", 

98 extra={"cmd": full, "version": version}, 

99 ) 

100 except KeyError: 

101 pass 

102 

103 try: 

104 tel: TelemetryProtocol = di.resolve(TelemetryProtocol) 

105 tel.event( 

106 "command_group_registered", {"command": full, "version": version} 

107 ) 

108 except KeyError: 

109 pass 

110 

111 return fn 

112 

113 return decorator 

114 

115 return with_sub 

116 

117 

118def dynamic_choices( 

119 callback: Callable[[], list[str]], 

120 *, 

121 case_sensitive: bool = True, 

122) -> Callable[[typer.Context, typer.models.ParameterInfo, str], list[str]]: 

123 """Creates a `Typer` completer from a callback function. 

124 

125 This factory function generates a completer that provides dynamic shell 

126 completion choices for a command argument or option. 

127 

128 Example: 

129 To provide dynamic completion for a `--user` option:: 

130 

131 def get_all_users() -> list[str]: 

132 return ["alice", "bob", "carol"] 

133 

134 @app.command() 

135 def delete( 

136 user: str = typer.Option( 

137 ..., 

138 autocompletion=dynamic_choices(get_all_users) 

139 ) 

140 ): 

141 ... 

142 

143 Args: 

144 callback (Callable[[], list[str]]): A no-argument function that returns 

145 a list of all possible choices. 

146 case_sensitive (bool): If True, prefix matching is case-sensitive. 

147 

148 Returns: 

149 A `Typer` completer function. 

150 """ 

151 

152 def completer( 

153 ctx: typer.Context, 

154 param: typer.models.ParameterInfo, 

155 incomplete: str, 

156 ) -> list[str]: 

157 """Filters the choices provided by the callback based on user input. 

158 

159 Args: 

160 ctx (typer.Context): The `Typer` command context. 

161 param (typer.models.ParameterInfo): The parameter being completed. 

162 incomplete (str): The current incomplete user input. 

163 

164 Returns: 

165 list[str]: A filtered list of choices that start with the 

166 `incomplete` string. 

167 """ 

168 items = callback() 

169 if case_sensitive: 

170 return [i for i in items if i.startswith(incomplete)] 

171 low = incomplete.lower() 

172 return [i for i in items if i.lower().startswith(low)] 

173 

174 return completer 

175 

176 

177__all__ = ["command_group", "dynamic_choices"]