Coverage for sources/appcore/cli.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 This module provides the core infrastructure for building command-line 

24 interfaces. It offers a comprehensive framework for creating CLI 

25 applications with rich presentation options, flexible output routing, and 

26 integrated logging capabilities. 

27 

28 Key Components 

29 ============== 

30 

31 Command Framework 

32 ----------------- 

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

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

35 * Rich integration with tyro for automatic argument parsing and help 

36 generation 

37 

38 Display and Output Control 

39 -------------------------- 

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

41 routing 

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

43 output 

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

45 * Rich terminal detection with colorization control 

46 

47 Example Usage 

48 ============= 

49 

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

51 

52 from appcore import cli, state 

53 

54 class MyDisplayOptions( cli.DisplayOptions ): 

55 format: str = 'table' 

56 

57 class MyGlobals( state.Globals ): 

58 display: MyDisplayOptions 

59 

60 class StatusCommand( cli.Command ): 

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

62 if isinstance( auxdata, MyGlobals ): 

63 format_val = auxdata.display.format 

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

65 

66 class InfoCommand( cli.Command ): 

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

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

69 

70 class MyApplication( cli.Application ): 

71 display: MyDisplayOptions = __.dcls.field( 

72 default_factory = MyDisplayOptions ) 

73 command: __.typx.Union[ 

74 __.typx.Annotated[ 

75 StatusCommand, 

76 _tyro.conf.subcommand( 'status', prefix_name = False ), 

77 ], 

78 __.typx.Annotated[ 

79 InfoCommand, 

80 _tyro.conf.subcommand( 'info', prefix_name = False ), 

81 ], 

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

83 

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

85 await self.command( auxdata ) 

86 

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

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

89 return MyGlobals( 

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

91''' 

92 

93 

94from . import __ 

95from . import exceptions as _exceptions 

96from . import inscription as _inscription 

97from . import preparation as _preparation 

98from . import state as _state 

99 

100 

101try: import rich 

102except ImportError as _error: # pragma: no cover 

103 raise _exceptions.DependencyAbsence( 'rich', 'CLI' ) from _error 

104else: del rich 

105try: import tomli_w 

106except ImportError as _error: # pragma: no cover 

107 raise _exceptions.DependencyAbsence( 'tomli-w', 'CLI' ) from _error 

108else: del tomli_w 

109try: import tyro as _tyro 

110except ImportError as _error: # pragma: no cover 

111 raise _exceptions.DependencyAbsence( 'tyro', 'CLI' ) from _error 

112 

113 

114_DisplayTargetMutex = _tyro.conf.create_mutex_group( required = False ) 

115_InscriptionTargetMutex = _tyro.conf.create_mutex_group( required = False ) 

116 

117 

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

119 ''' Target stream selection. ''' 

120 

121 Stdout = 'stdout' 

122 Stderr = 'stderr' 

123 

124 

125class DisplayOptions( __.immut.DataclassObject ): 

126 ''' Standardized display configuration for CLI applications. 

127 

128 Example:: 

129 

130 class MyDisplayOptions( DisplayOptions ): 

131 format: str = 'table' 

132 compact: bool = False 

133 ''' 

134 

135 colorize: __.typx.Annotated[ 

136 bool, 

137 _tyro.conf.arg( 

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

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

140 ] = True 

141 target_file: __.typx.Annotated[ 

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

143 _DisplayTargetMutex, 

144 _tyro.conf.DisallowNone, 

145 _tyro.conf.arg( help = "Render output to specified file." ), 

146 ] = None 

147 target_stream: __.typx.Annotated[ 

148 __.typx.Optional[ TargetStreams ], 

149 _DisplayTargetMutex, 

150 _tyro.conf.DisallowNone, 

151 _tyro.conf.arg( help = "Render output on stdout or stderr." ), 

152 ] = TargetStreams.Stdout 

153 assume_rich_terminal: __.typx.Annotated[ 

154 bool, 

155 _tyro.conf.arg( 

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

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

158 ] = False 

159 

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

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

162 if self.assume_rich_terminal: return self.colorize 

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

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

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

166 

167 async def provide_stream( 

168 self, exits: __.ctxl.AsyncExitStack 

169 ) -> __.typx.TextIO: 

170 ''' Provides target stream from options. ''' 

171 if self.target_file is not None: 

172 target_location = self.target_file.resolve( ) 

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

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

175 target_stream = self.target_stream or TargetStreams.Stderr 

176 match target_stream: 

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

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

179 

180 

181class InscriptionControl( __.immut.DataclassObject ): 

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

183 

184 level: __.typx.Annotated[ 

185 _inscription.Levels, _tyro.conf.arg( help = "Log verbosity." ) 

186 ] = 'info' 

187 presentation: __.typx.Annotated[ 

188 _inscription.Presentations, 

189 _tyro.conf.arg( help = "Log presentation mode (format)." ), 

190 ] = _inscription.Presentations.Plain 

191 target_file: __.typx.Annotated[ 

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

193 _InscriptionTargetMutex, 

194 _tyro.conf.DisallowNone, 

195 _tyro.conf.arg( help = "Log to specified file." ), 

196 ] = None 

197 target_stream: __.typx.Annotated[ 

198 __.typx.Optional[ TargetStreams ], 

199 _InscriptionTargetMutex, 

200 _tyro.conf.DisallowNone, 

201 _tyro.conf.arg( help = "Log to stdout or stderr." ), 

202 ] = TargetStreams.Stderr 

203 

204 def as_control( 

205 self, exits: __.ctxl.AsyncExitStack 

206 ) -> _inscription.Control: 

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

208 if self.target_file is not None: 

209 target_location = self.target_file.resolve( ) 

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

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

212 else: 

213 target_stream_ = self.target_stream or TargetStreams.Stderr 

214 match target_stream_: 

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

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

217 return _inscription.Control( 

218 mode = self.presentation, 

219 level = self.level, 

220 target = target_stream ) 

221 

222 

223class Command( 

224 __.immut.DataclassProtocol, __.typx.Protocol, 

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

226): 

227 ''' Standard interface for command implementations. 

228 

229 Example:: 

230 

231 class StatusCommand( Command ): 

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

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

234 ''' 

235 

236 async def __call__( self, auxdata: _state.Globals ) -> None: 

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

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

239 

240 @__.abc.abstractmethod 

241 async def execute( self, auxdata: _state.Globals ) -> None: 

242 ''' Executes command. ''' 

243 raise NotImplementedError # pragma: no cover 

244 

245 async def prepare( self, auxdata: _state.Globals ) -> _state.Globals: 

246 ''' Prepares session context. ''' 

247 return auxdata 

248 

249 

250class Application( 

251 __.immut.DataclassProtocol, __.typx.Protocol, 

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

253): 

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

255 

256 Example:: 

257 

258 class MyApplication( Application ): 

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: _state.Globals ) -> None: 

284 ''' Executes command. ''' 

285 raise NotImplementedError # pragma: no cover 

286 

287 async def prepare( self, exits: __.ctxl.AsyncExitStack ) -> _state.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 _preparation.prepare( exits, **nomargs )