Coverage for sources/appcore/cli/core.py: 98%

1# vim: set filetype=python fileencoding=utf-8: 

2# -*- coding: utf-8 -*- 

3 

4#============================================================================# 

5# # 

6# Licensed under the Apache License, Version 2.0 (the "License"); # 

7# you may not use this file except in compliance with the License. # 

8# You may obtain a copy of the License at # 

9# # 

10# http://www.apache.org/licenses/LICENSE-2.0 # 

11# # 

12# Unless required by applicable law or agreed to in writing, software # 

13# distributed under the License is distributed on an "AS IS" BASIS, # 

14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

15# See the License for the specific language governing permissions and # 

16# limitations under the License. # 

17# # 

18#============================================================================# 

19 

20 

21''' CLI foundation classes and interfaces. 

22 

23 Core infrastructure for building command-line interfaces. Comprehensive 

24 framework for creating CLI applications with rich presentation options, 

25 flexible output routing, and integrated logging capabilities. 

26 

27 Key Components 

28 ============== 

29 

30 Command Framework 

31 ----------------- 

32 * :class:`Command` - Abstract base class for CLI command implementations 

33 * :class:`Application` dataclass for command-line application configuration 

34 * Rich integration with Tyro for automatic argument parsing and help 

35 generation 

36 

37 Display and Output Control 

38 -------------------------- 

39 * :class:`DisplayOptions` - Configuration for output presentation and 

40 routing 

41 * :class:`InscriptionControl` - Configuration for logging and diagnostic 

42 output 

43 * Stream routing (stdout/stderr) and file output capabilities 

44 * Rich terminal detection with colorization control 

45 

46 Example Usage 

47 ============= 

48 

49 Basic CLI application with custom display options and subcommands:: 

50 

51 from appcore import cli, state 

52 

53 class MyDisplayOptions( cli.DisplayOptions ): 

54 format: str = 'table' 

55 

56 class MyGlobals( state.Globals ): 

57 display: MyDisplayOptions 

58 

59 class StatusCommand( cli.Command ): 

60 async def execute( self, auxdata: state.Globals ) -> None: 

61 if isinstance( auxdata, MyGlobals ): 

62 format_val = auxdata.display.format 

63 print( f"Status: Running (format: {format_val})" ) 

64 

65 class InfoCommand( cli.Command ): 

66 async def execute( self, auxdata: state.Globals ) -> None: 

67 print( f"App: {auxdata.application.name}" ) 

68 

69 class MyApplication( cli.Application ): 

70 display: MyDisplayOptions = __.dcls.field( 

71 default_factory = MyDisplayOptions ) 

72 command: __.typx.Union[ 

73 __.typx.Annotated[ 

74 StatusCommand, 

75 __.tyro.conf.subcommand( 'status', prefix_name = False ), 

76 ], 

77 __.typx.Annotated[ 

78 InfoCommand, 

79 __.tyro.conf.subcommand( 'info', prefix_name = False ), 

80 ], 

81 ] = __.dcls.field( default_factory = StatusCommand ) 

82 

83 async def execute( self, auxdata: state.Globals ) -> None: 

84 await self.command( auxdata ) 

85 

86 async def prepare( self, exits ) -> state.Globals: 

87 auxdata_base = await super( ).prepare( exits ) 

88 return MyGlobals( 

89 display = self.display, **auxdata_base.__dict__ ) 

90''' 

91 

92 

93from . import __ 

94 

95 

96_DisplayTargetMutex = __.tyro.conf.create_mutex_group( required = False ) 

97_InscriptionTargetMutex = __.tyro.conf.create_mutex_group( required = False ) 

98 

99 

100class TargetStreams( __.enum.Enum ): # TODO: Python 3.11: StrEnum 

101 ''' Target stream selection. ''' 

102 

103 Stdout = 'stdout' 

104 Stderr = 'stderr' 

105 

106 

107class DisplayOptions( __.immut.DataclassObject ): 

108 ''' Base display configuration for CLI applications. 

109 

110 Example:: 

111 

112 class MyDisplayOptions( DisplayOptions ): 

113 format: str = 'table' 

114 compact: bool = False 

115 ''' 

116 

117 colorize: __.typx.Annotated[ 

118 bool, 

119 __.tyro.conf.arg( 

120 aliases = ( '--ansi-sgr', ), 

121 help = "Enable colored output and terminal formatting." ), 

122 ] = True 

123 target_file: __.typx.Annotated[ 

124 __.typx.Optional[ __.Path ], 

125 _DisplayTargetMutex, 

126 __.tyro.conf.DisallowNone, 

127 __.tyro.conf.arg( help = "Render output to specified file." ), 

128 ] = None 

129 target_stream: __.typx.Annotated[ 

130 __.typx.Optional[ TargetStreams ], 

131 _DisplayTargetMutex, 

132 __.tyro.conf.DisallowNone, 

133 __.tyro.conf.arg( help = "Render output on stdout or stderr." ), 

134 ] = TargetStreams.Stdout 

135 assume_rich_terminal: __.typx.Annotated[ 

136 bool, 

137 __.tyro.conf.arg( 

138 aliases = ( '--force-tty', ), 

139 help = "Assume Rich terminal capabilities regardless of TTY." ), 

140 ] = False 

141 

142 def determine_colorization( self, stream: __.typx.TextIO ) -> bool: 

143 ''' Determines whether to use colorized output. ''' 

144 if self.assume_rich_terminal: return self.colorize 

145 if not self.colorize: return False 145 ↛ exitline 145 didn't return from function 'determine_colorization' because the return on line 145 wasn't executed

146 if __.os.environ.get( 'NO_COLOR' ): return False 146 ↛ exitline 146 didn't return from function 'determine_colorization' because the return on line 146 wasn't executed

147 return hasattr( stream, 'isatty' ) and stream.isatty( ) 

148 

149 async def provide_stream( 

150 self, exits: __.ctxl.AsyncExitStack 

151 ) -> __.typx.TextIO: 

152 ''' Provides target stream from options. ''' 

153 if self.target_file is not None: 

154 target_location = self.target_file.resolve( ) 

155 target_location.parent.mkdir( exist_ok = True, parents = True ) 

156 return exits.enter_context( target_location.open( 'w' ) ) 

157 target_stream = self.target_stream or TargetStreams.Stderr 

158 match target_stream: 

159 case TargetStreams.Stdout: return __.sys.stdout 

160 case TargetStreams.Stderr: return __.sys.stderr 

161 

162 

163class Globals( __.Globals ): 

164 ''' Application state with display options. ''' 

165 

166 display: DisplayOptions = __.dcls.field( default_factory = DisplayOptions ) 

167 

168 

169class InscriptionControl( __.immut.DataclassObject ): 

170 ''' Inscription (logging, debug prints) control. ''' 

171 

172 # TODO: Way to activate/deactivate all flavors (globally or per-address). 

173 # Format: --activate-all-flavors 

174 # Format: --activate-all-flavors-for <address> 

175 # Format: --deactivate-all-flavors 

176 # Format: --deactivate-all-flavors-for <address> 

177 # TODO: Way to activate particular flavors (globally or per-address). 

178 # Format: --active-flavor <name> 

179 # Format: --active-flavor <address>:<name> 

180 # TODO: Way to assign trace levels (globally or per-address). 

181 # Format: --trace-level <n> 

182 # Format: --trace-level <address>:<n> 

183 level: __.typx.Annotated[ 

184 __.inscription.Levels, __.tyro.conf.arg( help = "Log verbosity." ) 

185 ] = 'info' 

186 presentation: __.typx.Annotated[ 

187 __.inscription.Presentations, 

188 __.tyro.conf.arg( help = "Log presentation mode (format)." ), 

189 ] = __.inscription.Presentations.Plain 

190 target_file: __.typx.Annotated[ 

191 __.typx.Optional[ __.Path ], 

192 _InscriptionTargetMutex, 

193 __.tyro.conf.DisallowNone, 

194 __.tyro.conf.arg( help = "Log to specified file." ), 

195 ] = None 

196 target_stream: __.typx.Annotated[ 

197 __.typx.Optional[ TargetStreams ], 

198 _InscriptionTargetMutex, 

199 __.tyro.conf.DisallowNone, 

200 __.tyro.conf.arg( help = "Log to stdout or stderr." ), 

201 ] = TargetStreams.Stderr 

202 

203 def as_control( 

204 self, exits: __.ctxl.AsyncExitStack 

205 ) -> __.inscription.Control: 

206 ''' Produces compatible inscription control for appcore. ''' 

207 if self.target_file is not None: 

208 target_location = self.target_file.resolve( ) 

209 target_location.parent.mkdir( exist_ok = True, parents = True ) 

210 target_stream = exits.enter_context( target_location.open( 'w' ) ) 

211 else: 

212 target_stream_ = self.target_stream or TargetStreams.Stderr 

213 match target_stream_: 

214 case TargetStreams.Stdout: target_stream = __.sys.stdout 

215 case TargetStreams.Stderr: target_stream = __.sys.stderr 

216 return __.inscription.Control( 

217 mode = self.presentation, 

218 level = self.level, 

219 target = target_stream ) 

220 

221 

222class Command( 

223 __.immut.DataclassProtocol, __.typx.Protocol, 

224 decorators = ( __.typx.runtime_checkable, ), 

225): 

226 ''' Standard interface for command implementations. 

227 

228 Example:: 

229 

230 class StatusCommand( Command ): 

231 async def execute( self, auxdata: state.Globals ) -> None: 

232 print( f"Application: {auxdata.application.name}" ) 

233 ''' 

234 

235 async def __call__( self, auxdata: __.Globals ) -> None: 

236 ''' Prepares session context and executes command. ''' 

237 await self.execute( await self.prepare( auxdata ) ) 

238 

239 @__.abc.abstractmethod 

240 async def execute( self, auxdata: __.Globals ) -> None: 

241 ''' Executes command. ''' 

242 raise NotImplementedError # pragma: no cover 

243 

244 async def prepare( self, auxdata: __.Globals ) -> __.Globals: 

245 ''' Prepares session context. ''' 

246 return auxdata 

247 

248 

249class Application( 

250 __.immut.DataclassProtocol, __.typx.Protocol, 

251 decorators = ( __.typx.runtime_checkable, ), 

252): 

253 ''' Common infrastructure and standard interface for applications. 

254 

255 Example:: 

256 

257 class MyApplication( Application ): 

258 

259 display: DisplayOptions = __.dcls.field( 

260 default_factory = DisplayOptions ) 

261 

262 async def execute( self, auxdata: state.Globals ) -> None: 

263 print( f"Application: {auxdata.application.name}" ) 

264 ''' 

265 

266 configfile: __.typx.Annotated[ 

267 __.typx.Optional[ __.Path ], 

268 __.tyro.conf.arg( help = "Path to configuration file." ), 

269 ] = None 

270 environment: __.typx.Annotated[ 

271 bool, __.tyro.conf.arg( help = "Load environment from dotfiles?" ) 

272 ] = True 

273 inscription: InscriptionControl = __.dcls.field( 

274 default_factory = InscriptionControl ) 

275 

276 async def __call__( self ) -> None: 

277 ''' Prepares session context and executes command. ''' 

278 async with __.ctxl.AsyncExitStack( ) as exits: 

279 auxdata = await self.prepare( exits ) 

280 await self.execute( auxdata ) 

281 

282 @__.abc.abstractmethod 

283 async def execute( self, auxdata: __.Globals ) -> None: 

284 ''' Executes command. ''' 

285 raise NotImplementedError # pragma: no cover 

286 

287 async def prepare( self, exits: __.ctxl.AsyncExitStack ) -> __.Globals: 

288 ''' Prepares session context. ''' 

289 nomargs: __.NominativeArguments = dict( 

290 environment = self.environment, 

291 inscription = self.inscription.as_control( exits ) ) 

292 if self.configfile is not None: 

293 nomargs[ 'configfile' ] = self.configfile 

294 return await __.prepare( exits, **nomargs )