Coverage for sources/appcore/cli.py: 100%

70 statements  

« prev     ^ index     » next       coverage.py v7.10.7, created at 2025-09-26 19:13 +0000

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 

48 

49from . import __ 

50from . import exceptions as _exceptions 

51from . import inscription as _inscription 

52from . import preparation as _preparation 

53from . import state as _state 

54 

55 

56try: import rich 

57except ImportError as _error: # pragma: no cover 

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

59else: del rich 

60try: import tomli_w 

61except ImportError as _error: # pragma: no cover 

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

63else: del tomli_w 

64try: import tyro as _tyro 

65except ImportError as _error: # pragma: no cover 

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

67 

68 

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

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

71 

72 

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

74 ''' Target stream selection. ''' 

75 

76 Stdout = 'stdout' 

77 Stderr = 'stderr' 

78 

79 

80class DisplayOptions( __.immut.DataclassObject ): 

81 ''' Standardized display configuration for CLI applications. ''' 

82 

83 colorize: __.typx.Annotated[ 

84 bool, 

85 _tyro.conf.arg( 

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

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

88 ] = True 

89 target_file: __.typx.Annotated[ 

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

91 _DisplayTargetMutex, 

92 _tyro.conf.DisallowNone, 

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

94 ] = None 

95 target_stream: __.typx.Annotated[ 

96 __.typx.Optional[ TargetStreams ], 

97 _DisplayTargetMutex, 

98 _tyro.conf.DisallowNone, 

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

100 ] = TargetStreams.Stdout 

101 assume_rich_terminal: __.typx.Annotated[ 

102 bool, 

103 _tyro.conf.arg( 

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

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

106 ] = False 

107 

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

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

110 if self.assume_rich_terminal: 

111 return self.colorize 

112 return ( 

113 self.colorize 

114 and hasattr( stream, 'isatty' ) 

115 and stream.isatty( ) ) 

116 

117 async def provide_stream( 

118 self, exits: __.ctxl.AsyncExitStack 

119 ) -> __.typx.TextIO: 

120 ''' Provides target stream from options. ''' 

121 if self.target_file is not None: 

122 target_location = self.target_file.resolve( ) 

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

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

125 target_stream = self.target_stream or TargetStreams.Stderr 

126 match target_stream: 

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

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

129 

130 

131class InscriptionControl( __.immut.DataclassObject ): 

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

133 

134 level: __.typx.Annotated[ 

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

136 ] = 'info' 

137 presentation: __.typx.Annotated[ 

138 _inscription.Presentations, 

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

140 ] = _inscription.Presentations.Plain 

141 target_file: __.typx.Annotated[ 

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

143 _InscriptionTargetMutex, 

144 _tyro.conf.DisallowNone, 

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

146 ] = None 

147 target_stream: __.typx.Annotated[ 

148 __.typx.Optional[ TargetStreams ], 

149 _InscriptionTargetMutex, 

150 _tyro.conf.DisallowNone, 

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

152 ] = TargetStreams.Stderr 

153 

154 def as_control( 

155 self, exits: __.ctxl.AsyncExitStack 

156 ) -> _inscription.Control: 

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

158 if self.target_file is not None: 

159 target_location = self.target_file.resolve( ) 

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

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

162 else: 

163 target_stream_ = self.target_stream or TargetStreams.Stderr 

164 match target_stream_: 

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

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

167 return _inscription.Control( 

168 mode = self.presentation, 

169 level = self.level, 

170 target = target_stream ) 

171 

172 

173class Command( 

174 __.immut.DataclassProtocol, __.typx.Protocol, 

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

176): 

177 ''' Standard interface for command implementations. ''' 

178 

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

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

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

182 

183 @__.abc.abstractmethod 

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

185 ''' Executes command. ''' 

186 raise NotImplementedError # pragma: no cover 

187 

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

189 ''' Prepares session context. ''' 

190 return auxdata 

191 

192 

193class Application( 

194 __.immut.DataclassProtocol, __.typx.Protocol, 

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

196): 

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

198 

199 configfile: __.typx.Annotated[ 

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

201 _tyro.conf.arg( help = "Path to configuration file." ), 

202 ] = None 

203 environment: __.typx.Annotated[ 

204 bool, _tyro.conf.arg( help = "Load environment from dotfiles?" ) 

205 ] = True 

206 inscription: InscriptionControl = __.dcls.field( 

207 default_factory = InscriptionControl ) 

208 

209 async def __call__( self ) -> None: 

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

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

212 auxdata = await self.prepare( exits ) 

213 await self.execute( auxdata ) 

214 

215 @__.abc.abstractmethod 

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

217 ''' Executes command. ''' 

218 raise NotImplementedError # pragma: no cover 

219 

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

221 ''' Prepares session context. ''' 

222 nomargs: __.NominativeArguments = dict( 

223 environment = self.environment, 

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

225 if self.configfile is not None: 

226 nomargs[ 'configfile' ] = self.configfile 

227 return await _preparation.prepare( exits, **nomargs )