Coverage for sources/librovore/cli.py: 40%

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''' Command-line interface. '''

24from . import __

25from . import cacheproxy as _cacheproxy

26from . import exceptions as _exceptions

27from . import functions as _functions

28from . import interfaces as _interfaces

29from . import results as _results

30from . import server as _server

31from . import state as _state

34_scribe = __.acquire_scribe( __name__ )

37def intercept_errors( ) -> __.cabc.Callable[

38 [ __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ] ],

39 __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]

40]:

41 ''' Decorator for CLI handlers to intercept exceptions.

43 Catches Omnierror exceptions and renders them appropriately.

44 Other exceptions are logged and formatted simply.

45 '''

46 def decorator(

47 func: __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]

48 ) -> __.cabc.Callable[ ..., __.cabc.Awaitable[ None ] ]:

49 @__.funct.wraps( func )

50 async def wrapper(

51 self: __.typx.Any,

52 auxdata: _state.Globals,

53 display: __.DisplayTarget,

54 display_format: _interfaces.DisplayFormat,

55 *posargs: __.typx.Any,

56 **nomargs: __.typx.Any,

57 ) -> None:

58 stream = await display.provide_stream( )

59 try:

60 return await func(

61 self, auxdata, display, display_format,

62 *posargs, **nomargs )

63 except _exceptions.Omnierror as exc:

64 match display_format:

65 case _interfaces.DisplayFormat.JSON:

66 serialized = dict( exc.render_as_json( ) )

67 error_message = __.json.dumps( serialized, indent = 2 )

68 case _interfaces.DisplayFormat.Markdown:

69 lines = exc.render_as_markdown( )

70 error_message = '\n'.join( lines )

71 print( error_message, file = stream )

72 raise SystemExit( 1 ) from None

73 except Exception as exc:

74 _scribe.error( f"{func.__name__} failed: %s", exc )

75 match display_format:

76 case _interfaces.DisplayFormat.JSON:

77 error_data = {

78 "type": "unexpected_error",

79 "title": "Unexpected Error",

80 "message": str( exc ),

81 "suggestion": (

82 "Please report this issue if it persists." ),

83 }

84 error_message = __.json.dumps( error_data, indent = 2 )

85 case _interfaces.DisplayFormat.Markdown:

86 error_message = f"❌ Unexpected error: {exc}"

87 print( error_message, file = stream )

88 raise SystemExit( 1 ) from None

90 return wrapper

91 return decorator

94GroupByArgument: __.typx.TypeAlias = __.typx.Annotated[

95 __.typx.Optional[ str ],

96 __.tyro.conf.arg( help = __.access_doctab( 'group by argument' ) ),

97]

98PortArgument: __.typx.TypeAlias = __.typx.Annotated[

99 __.typx.Optional[ int ],

100 __.tyro.conf.arg( help = __.access_doctab( 'server port argument' ) ),

101]

102TermArgument: __.typx.TypeAlias = __.typx.Annotated[

103 __.tyro.conf.Positional[ str ],

104 __.tyro.conf.arg( help = __.access_doctab( 'term argument' ) ),

105]

106ResultsMax: __.typx.TypeAlias = __.typx.Annotated[

107 int,

108 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),

109]

110LocationArgument: __.typx.TypeAlias = __.typx.Annotated[

111 __.tyro.conf.Positional[ str ],

112 __.tyro.conf.arg( help = __.access_doctab( 'location argument' ) ),

113]

114TransportArgument: __.typx.TypeAlias = __.typx.Annotated[

115 __.typx.Optional[ str ],

116 __.tyro.conf.arg( help = __.access_doctab( 'transport argument' ) ),

117]

118

119

120_search_behaviors_default = _interfaces.SearchBehaviors( )

121_filters_default = __.immut.Dictionary[ str, __.typx.Any ]( )

122

123_MARKDOWN_OBJECT_LIMIT = 10

124_MARKDOWN_CONTENT_LIMIT = 200

125

126

127

128class _CliCommand(

129 __.immut.DataclassProtocol, __.typx.Protocol,

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

131):

132 ''' CLI command. '''

133

134 @__.abc.abstractmethod

135 def __call__(

136 self,

137 auxdata: _state.Globals,

138 display: __.DisplayTarget,

139 display_format: _interfaces.DisplayFormat,

140 ) -> __.cabc.Awaitable[ None ]:

141 ''' Executes command with global state. '''

142 raise NotImplementedError

143

144

145class DetectCommand(

146 _CliCommand, decorators = ( __.standard_tyro_class, ),

147):

148 ''' Detect which processors can handle a documentation source. '''

149

150 location: LocationArgument

151 genus: __.typx.Annotated[

152 _interfaces.ProcessorGenera,

153 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),

154 ]

155 processor_name: __.typx.Annotated[

156 __.typx.Optional[ str ],

157 __.tyro.conf.arg( help = "Specific processor to use." ),

158 ] = None

159

160 @intercept_errors( )

161 async def __call__(

162 self,

163 auxdata: _state.Globals,

164 display: __.DisplayTarget,

165 display_format: _interfaces.DisplayFormat,

166 ) -> None:

167 stream = await display.provide_stream( )

168 processor_name = (

169 self.processor_name if self.processor_name is not None

170 else __.absent )

171 result = await _functions.detect(

172 auxdata, self.location, self.genus,

173 processor_name = processor_name )

174 match display_format:

175 case _interfaces.DisplayFormat.JSON:

176 serialized = dict( result.render_as_json( ) )

177 output = __.json.dumps( serialized, indent = 2 )

178 case _interfaces.DisplayFormat.Markdown:

179 lines = result.render_as_markdown( reveal_internals = True )

180 output = '\n'.join( lines )

181 print( output, file = stream )

182

183

184class QueryInventoryCommand(

185 _CliCommand, decorators = ( __.standard_tyro_class, ),

186):

187 ''' Searches object inventory by name with fuzzy matching. '''

188

189 location: LocationArgument

190 term: TermArgument

191 details: __.typx.Annotated[

192 _interfaces.InventoryQueryDetails,

193 __.tyro.conf.arg(

194 help = __.access_doctab( 'query details argument' ) ),

195 ] = _interfaces.InventoryQueryDetails.Name

196 filters: __.typx.Annotated[

197 __.cabc.Mapping[ str, __.typx.Any ],

198 __.tyro.conf.arg( prefix_name = False ),

199 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )

200 search_behaviors: __.typx.Annotated[

201 _interfaces.SearchBehaviors,

202 __.tyro.conf.arg( prefix_name = False ),

203 ] = __.dcls.field(

204 default_factory = lambda: _interfaces.SearchBehaviors( ) )

205 results_max: __.typx.Annotated[

206 int,

207 __.tyro.conf.arg( help = __.access_doctab( 'results max argument' ) ),

208 ] = 5

209

210 @intercept_errors( )

211 async def __call__(

212 self,

213 auxdata: _state.Globals,

214 display: __.DisplayTarget,

215 display_format: _interfaces.DisplayFormat,

216 ) -> None:

217 stream = await display.provide_stream( )

218 result = await _functions.query_inventory(

219 auxdata,

220 self.location,

221 self.term,

222 search_behaviors = self.search_behaviors,

223 filters = self.filters,

224 results_max = self.results_max,

225 details = self.details )

226 match display_format:

227 case _interfaces.DisplayFormat.JSON:

228 output = __.json.dumps(

229 dict( result.render_as_json( ) ), indent = 2 )

230 case _interfaces.DisplayFormat.Markdown:

231 lines = result.render_as_markdown( reveal_internals = True )

232 output = '\n'.join( lines )

233 print( output, file = stream )

234

235

236class QueryContentCommand(

237 _CliCommand, decorators = ( __.standard_tyro_class, ),

238):

239 ''' Searches documentation content with relevance ranking and snippets. '''

240

241 location: LocationArgument

242 term: TermArgument

243 search_behaviors: __.typx.Annotated[

244 _interfaces.SearchBehaviors,

245 __.tyro.conf.arg( prefix_name = False ),

246 ] = __.dcls.field(

247 default_factory = lambda: _interfaces.SearchBehaviors( ) )

248 filters: __.typx.Annotated[

249 __.cabc.Mapping[ str, __.typx.Any ],

250 __.tyro.conf.arg( prefix_name = False ),

251 ] = __.dcls.field( default_factory = lambda: dict( _filters_default ) )

252 results_max: ResultsMax = 10

253 lines_max: __.typx.Annotated[

254 int,

255 __.tyro.conf.arg(

256 help = "Maximum number of lines to display per result." ),

257 ] = 40

258

259 @intercept_errors( )

260 async def __call__(

261 self,

262 auxdata: _state.Globals,

263 display: __.DisplayTarget,

264 display_format: _interfaces.DisplayFormat,

265 ) -> None:

266 stream = await display.provide_stream( )

267 result = await _functions.query_content(

268 auxdata, self.location, self.term,

269 search_behaviors = self.search_behaviors,

270 filters = self.filters,

271 results_max = self.results_max,

272 lines_max = self.lines_max )

273 match display_format:

274 case _interfaces.DisplayFormat.JSON:

275 output = __.json.dumps(

276 dict( result.render_as_json( ) ), indent = 2 )

277 case _interfaces.DisplayFormat.Markdown:

278 if isinstance( result, _results.ContentQueryResult ):

279 lines = result.render_as_markdown(

280 reveal_internals = True, lines_max = self.lines_max )

281 else:

282 lines = result.render_as_markdown(

283 reveal_internals = True )

284 output = '\n'.join( lines )

285 print( output, file = stream )

290class SurveyProcessorsCommand(

291 _CliCommand, decorators = ( __.standard_tyro_class, ),

292):

293 ''' List processors for specified genus and their capabilities. '''

294

295 genus: __.typx.Annotated[

296 _interfaces.ProcessorGenera,

297 __.tyro.conf.arg( help = "Processor genus (inventory or structure)." ),

298 ]

299 name: __.typx.Annotated[

300 __.typx.Optional[ str ],

301 __.tyro.conf.arg( help = "Name of processor to describe" ),

302 ] = None

303

304 @intercept_errors( )

305 async def __call__(

306 self,

307 auxdata: _state.Globals,

308 display: __.DisplayTarget,

309 display_format: _interfaces.DisplayFormat,

310 ) -> None:

311 stream = await display.provide_stream( )

312 nomargs: __.NominativeArguments = { 'genus': self.genus }

313 if self.name is not None: nomargs[ 'name' ] = self.name

314 result = await _functions.survey_processors( auxdata, **nomargs )

315 match display_format:

316 case _interfaces.DisplayFormat.JSON:

317 output = __.json.dumps(

318 dict( result.render_as_json( ) ), indent = 2 )

319 case _interfaces.DisplayFormat.Markdown:

320 lines = result.render_as_markdown( reveal_internals = False )

321 output = '\n'.join( lines )

322 print( output, file = stream )

323

324

325

326class ServeCommand(

327 _CliCommand, decorators = ( __.standard_tyro_class, ),

328):

329 ''' Starts MCP server. '''

330

331 port: PortArgument = None

332 transport: TransportArgument = None

333 extra_functions: __.typx.Annotated[

334 bool,

335 __.tyro.conf.arg(

336 help = "Enable extra functions (detect and survey-processors)." ),

337 ] = False

338 serve_function: __.typx.Callable[

339 [ _state.Globals ], __.cabc.Awaitable[ None ]

340 ] = _server.serve

341 async def __call__(

342 self,

343 auxdata: _state.Globals,

344 display: __.DisplayTarget,

345 display_format: _interfaces.DisplayFormat,

346 ) -> None:

347 nomargs: __.NominativeArguments = { }

348 if self.port is not None: nomargs[ 'port' ] = self.port

349 if self.transport is not None: nomargs[ 'transport' ] = self.transport

350 nomargs[ 'extra_functions' ] = self.extra_functions

351 await self.serve_function( auxdata, **nomargs )

352

353

354class Cli( __.immut.DataclassObject, decorators = ( __.simple_tyro_class, ) ):

355 ''' MCP server CLI. '''

356

357 display: __.DisplayTarget

358 display_format: __.typx.Annotated[

359 _interfaces.DisplayFormat,

360 __.tyro.conf.arg( help = "Output format for command results." ),

361 ] = _interfaces.DisplayFormat.Markdown

362 command: __.typx.Union[

363 __.typx.Annotated[

364 DetectCommand,

365 __.tyro.conf.subcommand( 'detect', prefix_name = False ),

366 ],

367 __.typx.Annotated[

368 QueryInventoryCommand,

369 __.tyro.conf.subcommand( 'query-inventory', prefix_name = False ),

370 ],

371 __.typx.Annotated[

372 QueryContentCommand,

373 __.tyro.conf.subcommand( 'query-content', prefix_name = False ),

374 ],

375 __.typx.Annotated[

376 SurveyProcessorsCommand,

377 __.tyro.conf.subcommand(

378 'survey-processors', prefix_name = False ),

379 ],

380 __.typx.Annotated[

381 ServeCommand,

382 __.tyro.conf.subcommand( 'serve', prefix_name = False ),

383 ],

384 ]

385 logfile: __.typx.Annotated[

386 __.typx.Optional[ str ],

387 __.ddoc.Doc( ''' Path to log capture file. ''' ),

388 ] = None

389

390 async def __call__( self ):

391 ''' Invokes command after library preparation. '''

392 nomargs = self.prepare_invocation_args( )

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

394 auxdata = await _prepare( exits = exits, **nomargs )

395 from . import xtnsmgr

396 await xtnsmgr.register_processors( auxdata )

397 await self.command(

398 auxdata = auxdata,

399 display = self.display,

400 display_format = self.display_format )

401

402 def prepare_invocation_args(

403 self,

404 ) -> __.cabc.Mapping[ str, __.typx.Any ]:

405 ''' Prepares arguments for initial configuration. '''

406 args: dict[ str, __.typx.Any ] = dict(

407 environment = True,

408 logfile = self.logfile,

409 )

410 return args

411

412

413def execute( ) -> None:

414 ''' Entrypoint for CLI execution. '''

415 config = (

416 __.tyro.conf.HelptextFromCommentsOff,

417 )

418 with __.warnings.catch_warnings( ):

419 __.warnings.filterwarnings(

420 'ignore',

421 message = r'Mutable type .* is used as a default value.*',

422 category = UserWarning,

423 module = 'tyro.constructors._struct_spec_dataclass' )

424 try: __.asyncio.run( __.tyro.cli( Cli, config = config )( ) )

425 except SystemExit: raise

426 except BaseException as exc:

427 __.report_exceptions( exc, _scribe )

428 raise SystemExit( 1 ) from None

441async def _prepare(

442 environment: __.typx.Annotated[

443 bool,

444 __.ddoc.Doc( ''' Whether to configure environment. ''' )

445 ],

446 exits: __.typx.Annotated[

447 __.ctxl.AsyncExitStack,

448 __.ddoc.Doc( ''' Exit stack for resource management. ''' )

449 ],

450 logfile: __.typx.Annotated[

451 __.typx.Optional[ str ],

452 __.ddoc.Doc( ''' Path to log capture file. ''' )

453 ],

454) -> __.typx.Annotated[

455 _state.Globals,

456 __.ddoc.Doc( ''' Configured global state. ''' )

457]:

458 ''' Configures application based on arguments. '''

459 nomargs: __.NominativeArguments = {

460 'environment': environment,

461 'exits': exits,

462 }

463 if logfile:

464 logfile_p = __.Path( logfile ).resolve( )

465 ( logfile_p.parent ).mkdir( parents = True, exist_ok = True )

466 logstream = exits.enter_context( logfile_p.open( 'w' ) )

467 inscription = __.appcore.inscription.Control(

468 level = 'debug', target = logstream )

469 nomargs[ 'inscription' ] = inscription

470 auxdata = await __.appcore.prepare( **nomargs )

471 content_cache, probe_cache, robots_cache = _cacheproxy.prepare( auxdata )

472 return _state.Globals(

473 application = auxdata.application,

474 configuration = auxdata.configuration,

475 directories = auxdata.directories,

476 distribution = auxdata.distribution,

477 exits = auxdata.exits,

478 content_cache = content_cache,

479 probe_cache = probe_cache,

480 robots_cache = robots_cache )