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
« 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 -*-
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#============================================================================#
21''' CLI foundation classes and interfaces.
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.
28 Key Components
29 ==============
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
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'''
49from . import __
50from . import exceptions as _exceptions
51from . import inscription as _inscription
52from . import preparation as _preparation
53from . import state as _state
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
69_DisplayTargetMutex = _tyro.conf.create_mutex_group( required = False )
70_InscriptionTargetMutex = _tyro.conf.create_mutex_group( required = False )
73class TargetStreams( __.enum.Enum ): # TODO: Python 3.11: StrEnum
74 ''' Target stream selection. '''
76 Stdout = 'stdout'
77 Stderr = 'stderr'
80class DisplayOptions( __.immut.DataclassObject ):
81 ''' Standardized display configuration for CLI applications. '''
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
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( ) )
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
131class InscriptionControl( __.immut.DataclassObject ):
132 ''' Inscription (logging, debug prints) control. '''
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
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 )
173class Command(
174 __.immut.DataclassProtocol, __.typx.Protocol,
175 decorators = ( __.typx.runtime_checkable, ),
176):
177 ''' Standard interface for command implementations. '''
179 async def __call__( self, auxdata: _state.Globals ) -> None:
180 ''' Prepares session context and executes command. '''
181 await self.execute( await self.prepare( auxdata ) )
183 @__.abc.abstractmethod
184 async def execute( self, auxdata: _state.Globals ) -> None:
185 ''' Executes command. '''
186 raise NotImplementedError # pragma: no cover
188 async def prepare( self, auxdata: _state.Globals ) -> _state.Globals:
189 ''' Prepares session context. '''
190 return auxdata
193class Application(
194 __.immut.DataclassProtocol, __.typx.Protocol,
195 decorators = ( __.typx.runtime_checkable, ),
196):
197 ''' Common infrastructure and standard interface for applications. '''
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 )
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 )
215 @__.abc.abstractmethod
216 async def execute( self, auxdata: _state.Globals ) -> None:
217 ''' Executes command. '''
218 raise NotImplementedError # pragma: no cover
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 )