87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import sys
|
||
|
|
|
||
|
|
from typing import TYPE_CHECKING, Any, override
|
||
|
|
|
||
|
|
from .App import App as Base
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
|
||
|
|
from argparse import ArgumentParser
|
||
|
|
|
||
|
|
from .ExecContext import ExecContext
|
||
|
|
|
||
|
|
class ExecApp(Base): # export
|
||
|
|
|
||
|
|
def __init__(self, **kwargs: Any) -> None:
|
||
|
|
super().__init__(**kwargs)
|
||
|
|
self.__opt_interactive: bool | None = None
|
||
|
|
self.__opt_verbose: bool | None = None
|
||
|
|
self.__exec_context: ExecContext | None = None
|
||
|
|
|
||
|
|
@override
|
||
|
|
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
|
||
|
|
if self.__exec_context is not None:
|
||
|
|
await self.__exec_context.close()
|
||
|
|
self.__exec_context = None
|
||
|
|
# -- Close the app (async runner, event loop) after the exec context,
|
||
|
|
# which may still need them.
|
||
|
|
await super().__aexit__(exc_type, exc, tb)
|
||
|
|
|
||
|
|
@override
|
||
|
|
def _add_arguments(self, parser: ArgumentParser) -> None:
|
||
|
|
super()._add_arguments(parser)
|
||
|
|
parser.add_argument(
|
||
|
|
'--interactive',
|
||
|
|
choices = ['true', 'false', 'auto'],
|
||
|
|
default = 'true',
|
||
|
|
help = 'Wait for user input or try to proceed unattended',
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
'--verbose',
|
||
|
|
action = 'store_true',
|
||
|
|
default = False,
|
||
|
|
help = "Be verbose on stderr about what's being done on the distro level",
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
'--target', default = 'local', help = 'Run commands on this host'
|
||
|
|
)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def interactive(self) -> bool:
|
||
|
|
if self.__opt_interactive is None:
|
||
|
|
match self.args.interactive:
|
||
|
|
case 'true':
|
||
|
|
self.__opt_interactive = True
|
||
|
|
case 'false':
|
||
|
|
self.__opt_interactive = False
|
||
|
|
case 'auto':
|
||
|
|
self.__opt_interactive = sys.stdin.isatty()
|
||
|
|
case _:
|
||
|
|
raise ValueError(
|
||
|
|
f'Unknown --interactive value: {self.args.interactive}'
|
||
|
|
)
|
||
|
|
# Not logically possible to fail, but this keeps pyright happy
|
||
|
|
assert self.__opt_interactive is not None
|
||
|
|
return self.__opt_interactive
|
||
|
|
|
||
|
|
@property
|
||
|
|
def verbose(self) -> bool:
|
||
|
|
if self.__opt_verbose is None:
|
||
|
|
self.__opt_verbose = self.args.verbose
|
||
|
|
# Not logically possible to fail, but this keeps pyright happy
|
||
|
|
assert self.__opt_verbose is not None
|
||
|
|
return self.__opt_verbose
|
||
|
|
|
||
|
|
@property
|
||
|
|
def exec_context(self) -> ExecContext:
|
||
|
|
if self.__exec_context is None:
|
||
|
|
from .ExecContext import ExecContext
|
||
|
|
|
||
|
|
self.__exec_context = ExecContext.create(
|
||
|
|
self.args.target,
|
||
|
|
interactive = self.interactive,
|
||
|
|
verbose_default = self.verbose,
|
||
|
|
)
|
||
|
|
return self.__exec_context
|