diff --git a/README.md b/README.md index bd69dae..999553a 100644 --- a/README.md +++ b/README.md @@ -41,9 +41,9 @@ During an *interactive session*, you can set the application in the current context. ```python # import other modules -import alphaconf.interactive -alphaconf.interactive.mount() -alphaconf.interactive.load_configuration_file('path') +import alphaconf +alphaconf.load_configuration_file('path') +alphaconf.initialize() ``` Check the [DEMO](./demo.ipynb) for more examples. @@ -162,12 +162,8 @@ alphaconf.invoke.run(__name__, ns) Replace `invoke` with alphaconf and plumbum. ## Way to 1.0 -- Make argument parsing separate from the core -- Compare `plumbum` and `invoke` for building scripts +- replace toml with tomllib (standard since 3.11) - Secret handling and encryption -- Run a specific function `alphaconf my.module.main`: - find functions and inject args -- Install completions for bash `alphaconf --install-autocompletion` [OmegaConf]: https://omegaconf.readthedocs.io/ [pydantic]: https://docs.pydantic.dev/latest/ diff --git a/alphaconf/__init__.py b/alphaconf/__init__.py index c05cc25..655872a 100644 --- a/alphaconf/__init__.py +++ b/alphaconf/__init__.py @@ -1,10 +1,10 @@ +import logging import re -import warnings -from typing import Callable, MutableSequence, Optional, Sequence, TypeVar, Union +from typing import Callable, MutableSequence, Optional, TypeVar from .frozendict import frozendict # noqa: F401 (expose) -from .internal.application import Application from .internal.configuration import Configuration +from .internal.load_file import read_configuration_file __doc__ = """AlphaConf @@ -45,49 +45,53 @@ """The global configuration""" setup_configuration = _global_configuration.setup_configuration - -_application: Optional[Application] = None get = _global_configuration.get +__initialized = False + + +def load_configuration_file(path: str): + """Read a configuration file and add it to the context configuration""" + config = read_configuration_file(path) + logging.debug('Loading configuration from path: %s', path) + setup_configuration(config) -def set_application(app: Application) -> None: - """Setup the application globally +def select_configuration(name, key): + pass # TODO - This loads the configuration and initializes the application. - The function may raise ExitApplication. - """ - global _application, get - if _application is app: + +def initialize( + app_name: str = '', + setup_logging: bool = True, + load_dotenv: Optional[bool] = None, + force: bool = False, +): + """Initialize the application and setup configuration""" + global __initialized + if __initialized and not force: + logging.info("The application is already initialized", stack_info=True) return - if _application is not None: - _application.log.info("Another application will be loaded") - _application = app - get = app.configuration.get + __initialized = True + + # load from dotenv + from .internal.dotenv_vars import try_dotenv + try_dotenv(load_dotenv=load_dotenv) -def run( - main: Callable[[], T], - arguments: Union[bool, Sequence[str]] = True, - *, - should_exit: bool = True, - app: Optional[Application] = None, - **config, -) -> Optional[T]: - """Run this application (deprecated) + # load the application + from .internal import application as app - If an application is not given, a new one will be created with configuration properties - taken from the config. Also, by default logging is set up. + configurations = app.get_configurations( + app_name=app_name, default_configuration=_global_configuration.c + ) + _global_configuration._merge(configurations) - :param main: The main function to call - :param arguments: List of arguments (default: True to read sys.argv) - :param should_exit: Whether an exception should sys.exit (default: True) - :param config: Arguments passed to Application.__init__() and Application.setup_configuration() - :return: The result of main - """ - warnings.warn("use alphaconf.cli.run directly", DeprecationWarning) - from .cli import run + # setup logging + if setup_logging: + from . import get, logging_util - return run(main, arguments, should_exit=should_exit, app=app, **config) + logging_util.setup_application_logging(get('logging', default=None)) + logging.debug('Application initialized') ####################################### @@ -156,4 +160,4 @@ def __alpha_configuration(): # Initialize configuration __alpha_configuration() -__all__ = ["get", "setup_configuration", "set_application", "Application", "frozendict"] +__all__ = ["get", "setup_configuration", "frozendict"] diff --git a/alphaconf/cli.py b/alphaconf/cli.py index 43b4e3a..877e624 100644 --- a/alphaconf/cli.py +++ b/alphaconf/cli.py @@ -1,46 +1,82 @@ import sys +import logging +import argparse from typing import Any, Callable, Optional, Sequence, TypeVar, Union from omegaconf import MissingMandatoryValue, OmegaConf -from . import set_application -from .internal.application import Application -from .internal.arg_parser import Action, ArgumentError, ExitApplication +from . import initialize, setup_configuration +from .internal.load_file import read_configuration_file T = TypeVar('T') -__all__ = ["run"] +__all__ = ["run", "Application"] +log = logging.getLogger(__name__) -class CommandAction(Action): - pass # TODO just read a function name, parse rest with this one - - -class CliApplication(Application): - commands: dict[str, Callable[[], Any]] - - def __init__(self, *, name: str | None = None, **properties) -> None: - super().__init__(name=name, **properties) - self.commands = {} - - def command(self, name=None, inject=False): - def register_command(func): - reg_name = name or func.__name__ - if inject: - from .inject import inject_auto - - func = inject_auto()(func) - self.commands[reg_name] = func - - return register_command - - def _run(self): - # TODO - for cmd in self.commands.values(): - return cmd() - return None - - def run(self, **config): - return run(self._run, **config, app=self) +class ConfigAction(argparse.Action): + def __call__(self, parser, namespace, values, option_string=None): + print(option_string, values) + if option_string: + config = read_configuration_file(values) + else: + config = OmegaConf.create(values[0]) + setup_configuration(config) + + +class SelectConfigAction(ConfigAction): + def __call__(self, parser, namespace, values, option_string=None): + key, value = values.split('=') # XXX _split(value) + value = value or 'default' + arg = "{key}=${{oc.select:base.{key}.{value}}}".format(key=key, value=value) + return super().__call__(parser, namespace, [arg], option_string) + + +class ShowConfigurationAction(argparse.Action): + def __call__(self, parser, namespace, values, option_string=None): + from . import _global_configuration + + config = _global_configuration + print(config.c) + parser.exit() + + +def parser_create(method, add_arguments=True, **args): + from .internal import application + + args.setdefault('prog', args.pop('name', None) or application.get_current_application_name()) + if method: + args.setdefault('description', method.__doc__) + args.setdefault('epilog', 'powered by alphaconf') + parser = argparse.ArgumentParser(**args) + if add_arguments: + parser_add_arguments(parser) + return parser + + +def parser_add_arguments(parser: argparse.ArgumentParser): + parser.add_argument( + '--config', + '--config-file', + '-f', + action=ConfigAction, + metavar="path", + help="load a configuration file", + ) + parser.add_argument( + '--select', + action=SelectConfigAction, + metavar="key=base_template", + help="select a configuration template", + ) + parser.add_argument( + '--configuration', + '-C', + nargs=0, + action=ShowConfigurationAction, + help="show the configuration", + ) + # TODO does not support -x a=5 -y b=5 + parser.add_argument('key=value', nargs='*', action=ConfigAction, help="add configuration") def run( @@ -48,11 +84,10 @@ def run( arguments: Union[bool, Sequence[str]] = True, *, should_exit: bool = True, - app: Optional[Application] = None, setup_logging: bool = True, **config, ) -> Optional[T]: - """Run this application + """Run a function/application If an application is not given, a new one will be created with configuration properties taken from the config. Also, by default logging is set up. @@ -63,45 +98,28 @@ def run( :param config: Arguments passed to Application.__init__() and Application.setup_configuration() :return: The result of main """ - # Create the application if needed - if app is None: - properties = { - k: config.pop(k) - for k in ['name', 'version', 'description', 'short_description'] - if k in config - } - # if we don't have a description, get it from the function's docs - if 'description' not in properties and main.__doc__: - description = main.__doc__.strip().split('\n', maxsplit=1) - if 'short_description' not in properties: - properties['short_description'] = description[0] - if len(description) > 1: - import textwrap - - properties['description'] = description[0] + '\n' + textwrap.dedent(description[1]) - else: - properties['description'] = properties['short_description'] - app = Application(**properties) - log = app.log - - # Setup the application + from . import get, _global_configuration + + arg_parser = parser_create(main, **config, exit_on_error=should_exit) try: + initialize(app_name=arg_parser.prog, setup_logging=False, force=True) if arguments is True: arguments = sys.argv[1:] if not isinstance(arguments, list): arguments = [] - app.setup_configuration(arguments=arguments, **config) - set_application(app) - configuration = app.configuration + args = arg_parser.parse_args(arguments) + # args = arg_parser.parse_intermixed_args(arguments) # XXX NOT WHAT I WANT + print(args) # TODO if setup_logging: from .logging_util import setup_application_logging - setup_application_logging(configuration.get('logging', default=None)) + setup_application_logging(get('logging', default=None)) except MissingMandatoryValue as e: log.error(e) if should_exit: sys.exit(99) raise + """ except ArgumentError as e: log.error(e) if should_exit: @@ -112,14 +130,15 @@ def run( if should_exit: sys.exit() return None + """ # Run the application - if configuration.get('testing', bool, default=False): - log.info('Testing (%s: %s)', app.name, main.__qualname__) + if get('testing', bool, default=False): + log.info('Testing (%s: %s)', arg_parser.prog, main.__qualname__) return None try: - log.info('Start (%s: %s)', app.name, main.__qualname__) - for missing_key in OmegaConf.missing_keys(configuration.c): + log.info('Start (%s: %s)', arg_parser.prog, main.__qualname__) + for missing_key in OmegaConf.missing_keys(_global_configuration.c): log.warning('Missing configuration key: %s', missing_key) result = main() if result is None: diff --git a/alphaconf/inject.py b/alphaconf/inject.py index 26de30d..5aa69b7 100644 --- a/alphaconf/inject.py +++ b/alphaconf/inject.py @@ -76,7 +76,7 @@ def do_inject(func): return do_inject -def inject_auto(*, prefix: str = "", ignore: set = set()): +def inject_auto(*, prefix: str = "", ignore: set[str] = set()): """Inject automatically all paramters""" if prefix and not prefix.endswith("."): prefix += "." diff --git a/alphaconf/interactive.py b/alphaconf/interactive.py deleted file mode 100644 index 53216a7..0000000 --- a/alphaconf/interactive.py +++ /dev/null @@ -1,31 +0,0 @@ -import logging -from typing import List - -from . import set_application -from .internal.application import Application -from .internal.load_file import read_configuration_file - -__doc__ = """Helpers for interactive applications like ipython.""" -__all__ = ['mount', 'read_configuration_file', 'load_configuration_file'] - -application = Application(name="interactive") - - -def mount(configuration_paths: List[str] = [], setup_logging: bool = True): - """Mount the interactive application and setup configuration""" - application.setup_configuration(configuration_paths=configuration_paths) - set_application(application) - if setup_logging: - from . import logging_util - - logging_util.setup_application_logging( - application.configuration.get('logging', default=None) - ) - logging.info('Mounted interactive application') - - -def load_configuration_file(path: str): - """Read a configuration file and add it to the context configuration""" - config = read_configuration_file(path) - logging.debug('Loading configuration from path: %s', path) - application.configuration.setup_configuration(config) diff --git a/alphaconf/internal/application.py b/alphaconf/internal/application.py index b4e3691..b829eb1 100644 --- a/alphaconf/internal/application.py +++ b/alphaconf/internal/application.py @@ -7,19 +7,135 @@ from omegaconf import DictConfig, OmegaConf -from . import arg_parser, load_file +from . import load_file from .configuration import Configuration -class Application: +def get_current_application_name() -> str: + """Find the default name from sys.argv""" + name = os.path.basename(sys.argv[0]) + if name.endswith('.py'): + name = name[:-3] + if name == '__main__': + # executing a module using python -m + name = os.path.basename(os.path.dirname(sys.argv[0])) + return name + + +def _application_configuration(app_name: str = '', version: str = '') -> DictConfig: + """Get the application configuration key""" + return OmegaConf.create( + { + 'application': { + 'name': app_name or get_current_application_name(), + 'version': version, + 'uuid': str(uuid.uuid4()), + }, + } + ) + + +def possible_configuration_paths(app_name: str = '') -> Iterable[str]: + """List of paths where to find configuration files""" + name = app_name or get_current_application_name() + is_windows = sys.platform.startswith('win') + for path in [ + '$APPDATA/{}' if is_windows else '/etc/{}', + '$LOCALAPPDATA/{}' if is_windows else '', + '$HOME/.{}', + '$HOME/.config/{}', + '$PWD/{}', + ]: + path = path and os.path.expandvars(path) + if path and '$' not in path: + for ext in load_file.SUPPORTED_EXTENSIONS: + yield path.format(f"{name}.{ext}") + + +def get_configurations( + default_configuration: DictConfig, # TODO use this or get the global config + app_name: str = '', + configuration_paths: Iterable[str] = [], + env_prefixes: Union[bool, Iterable[str]] = True, +) -> Iterable[DictConfig]: + """List of all configurations that can be loaded automatically + + - Global configuration + - The app configuration + - Read file defined in PYTHON_ALPHACONF + - Reads existing files from possible configuration paths + - Reads environment variables based on given prefixes + + :param env_prefixes: Prefixes of environment variables to load + :return: OmegaConf configurations (to be merged) + """ + log = logging.getLogger() + # App + yield default_configuration + if app_name: + yield _application_configuration(app_name) + log.debug('Loading default and app configurations') + # Read files + env_configuration_path = os.environ.get('PYTHON_ALPHACONF') or '' + for path in itertools.chain( + [env_configuration_path], + possible_configuration_paths(app_name=app_name) if app_name else [], + configuration_paths, + ): + if not os.path.isfile(path): + continue + log.debug('Load configuration from %s', path) + yield load_file.read_configuration_file(path) + # Environment + prefixes: Optional[Tuple[str, ...]] + if env_prefixes is True: + log.debug('Detecting accepted env prefixes') + default_keys = {str(k) for k in default_configuration} + prefixes = tuple( + k.upper() + '_' + for k in default_keys + if k not in ('base', 'python') and not k.startswith('_') + ) + elif isinstance(env_prefixes, Iterable): + prefixes = tuple(env_prefixes) + else: + prefixes = None + if prefixes: + log.debug('Loading env configuration from prefixes %s', prefixes) + yield from_environ(default_configuration, prefixes) + + +def from_environ(c: DictConfig, prefixes: Iterable[str]) -> DictConfig: + """Load environment variables into a dict configuration""" + from yaml.error import YAMLError # type: ignore + + trans = str.maketrans('_', '.', '"\\=') + prefixes = tuple(prefixes) + dotlist = [ + (name.lower().translate(trans).strip('.'), value) + for name, value in os.environ.items() + if name.startswith(prefixes) + ] + conf = OmegaConf.create({}) + for name, value in dotlist: + name = Configuration._find_name(name.split('.'), c) + try: + conf.merge_with_dotlist([f"{name}={value}"]) + except YAMLError: + # if cannot load the value as a dotlist, just add the string + OmegaConf.update(conf, name, value) + return conf + + +class DeprecatedApplication: # TODO remove """An application description""" log = logging.getLogger('alphaconf') __config: Optional[Configuration] = None __name: str properties: MutableMapping[str, str] - argument_parser: arg_parser.ArgumentParser - parsed: Optional[arg_parser.ParseResult] = None + # argument_parser: old_arg_parser.ArgumentParser + # parsed: Optional[old_arg_parser.ParseResult] = None def __init__( self, @@ -40,11 +156,11 @@ def __init__( self.properties = properties self.argument_parser = self._build_argument_parser() - def _build_argument_parser(self) -> arg_parser.ArgumentParser: + def _build_argument_parser(self): # -> old_arg_parser.ArgumentParser: from .. import _global_configuration - p = arg_parser.ArgumentParser(_global_configuration.helpers) - arg_parser.configure_parser(p, app=self) + p = old_arg_parser.ArgumentParser(_global_configuration.helpers) + old_arg_parser.configure_parser(p, app=self) return p @staticmethod @@ -190,7 +306,7 @@ def _handle_parsed_result(self): if self.parsed.result: return self.parsed.result.run(self) if self.parsed.rest: - raise arg_parser.ArgumentError(f"Too many arguments {self.parsed.rest}") + raise old_arg_parser.ArgumentError(f"Too many arguments {self.parsed.rest}") return None def masked_configuration( diff --git a/alphaconf/internal/arg_parser.py b/alphaconf/internal/arg_parser.py deleted file mode 100644 index c140248..0000000 --- a/alphaconf/internal/arg_parser.py +++ /dev/null @@ -1,320 +0,0 @@ -import itertools -from typing import Dict, Iterable, List, Mapping, Optional, Tuple, Type, Union, cast - -from omegaconf import DictConfig, OmegaConf - - -def _split(value: str, char: str = "=") -> Tuple[str, Optional[str]]: - vs = value.split(char, 1) - if len(vs) < 2: - return vs[0], None - return vs[0], vs[1] - - -class ExitApplication(BaseException): - """Signal to exit the application normally""" - - pass - - -class ArgumentError(RuntimeError): - """Argument parsing error""" - - def __init__(self, message: str, *args: object, arg=None) -> None: - if arg: - message = f"{arg}: {message}" - super().__init__(message, *args) - - -class Action: - """Action for parsing""" - - def __init__(self, *, metavar: Optional[str] = None, help: Optional[str] = None) -> None: - self.metavar = metavar - self.help = help - self.has_arg = bool(metavar) - - def check_argument(self, value: Optional[str]) -> Optional[str]: - if not value and self.metavar: - return "Required value" - return None - - def handle(self, result: "ParseResult", value: Optional[str]) -> str: - if result.result: - return "Result is already set" - result.result = self - return 'stop' - - def run(self, app): - raise ArgumentError(f"Cannot execute action {self}") - - def __str__(self) -> str: - return type(self).__name__ - - -class ShowConfigurationAction(Action): - """Show configuration action""" - - def run(self, app): - output = OmegaConf.to_yaml(app.masked_configuration()) - print(output) - raise ExitApplication - - -class HelpAction(Action): - """Help action""" - - def run(self, app): - app.print_help() - raise ExitApplication - - -class VersionAction(Action): - """Version action""" - - def run(self, app): - prog = app.name - p = app.properties - version = p.get('version') - if version: - print(f"{prog} {version}") - else: - print(prog) - desc = p.get('short_description') - if desc: - print(desc) - raise ExitApplication - - -class ConfigurationAction(Action): - """Configuration action""" - - def check_argument(self, value): - if self.metavar and '=' in self.metavar and '=' not in value: - return 'Argument should be in format %s' % self.metavar - return super().check_argument(value) - - def handle(self, result, value): - result._add_config(value) - - -class ConfigurationFileAction(ConfigurationAction): - """Load configuration file action""" - - def check_argument(self, value): - if not value: - return 'Missing filename for configuration file' - return None - - def handle(self, result, value): - result._add_config(OmegaConf.load(value)) - - -class ConfigurationSelectAction(ConfigurationAction): - """oc.select configuration action""" - - def check_argument(self, value): - return Action.check_argument(self, value) - - def handle(self, result, value): - key, value = _split(value) - value = value or 'default' - arg = "{key}=${{oc.select:base.{key}.{value}}}".format(key=key, value=value) - return super().handle(result, arg) - - -class ParseResult: - """The result of argument parsing""" - - result: Optional[Action] - rest: List[str] - _config: List[Union[str, DictConfig]] - - def __init__(self) -> None: - """Initialize the result""" - self.result = None - self.rest = [] - self._config = [] - - def _add_config(self, value: Union[List[str], DictConfig, Dict, str]): - """Add a configuration item""" - if isinstance(value, list): - self._config.extend(value) - return - elif isinstance(value, DictConfig): - pass - elif isinstance(value, dict): - value = OmegaConf.create(value) - elif isinstance(value, str): - pass - else: - raise ArgumentError(f"Invalid configuration type {type(value)}") - self._config.append(value) - - def configurations(self) -> Iterable[DictConfig]: - """List parsed configuration dicts""" - configuration_list = self._config - if not configuration_list: - return - for typ, conf in itertools.groupby(configuration_list, type): - if issubclass(typ, DictConfig): - yield from cast(Iterable[DictConfig], conf) - else: - yield OmegaConf.from_dotlist(list(cast(Iterable[str], conf))) - - def __repr__(self) -> str: - return f"(result={self.result}, config={self._config}, rest={self.rest})" - - -class ArgumentParser: - """Parses arguments for alphaconf""" - - _opt_actions: Dict[str, Action] - _pos_actions: List[Action] - help_messages: Mapping[str, str] - - def __init__(self, help_messages: Mapping[str, str] = {}) -> None: - self._opt_actions = {} - self._pos_actions = [] - self.help_messages = help_messages or {} - - def parse_args(self, arguments: List[str]) -> ParseResult: - """Parse the argument""" - result = ParseResult() - arguments = list(arguments) - arguments.reverse() - while arguments: - arg = arguments.pop() - if arg == '--': - break - value = None - is_opt = arg.startswith('-') - if is_opt and '=' in arg: - # arg is -xxx=yyy, split it - arg, value = _split(arg) - if not arg.startswith('--') and len(arg) != 2: - raise ArgumentError("Short option must be alone with a value", arg=arg) - if is_opt: - # parse option arguments - action = self._opt_actions.get(arg) - if not action: - raise ArgumentError('Unrecognized option', arg=arg) - if value is None and action.has_arg: - if not arguments: - raise ArgumentError(f"No more arguments to read {action.metavar}", arg=arg) - value = arguments.pop() - elif value is not None and not action.has_arg: - raise ArgumentError("Action has no arguments", arg=arg) - error = action.check_argument(value) - if error: - raise ArgumentError(error, arg=arg) - action_result = action.handle(result, value) - else: - # parse positional arguments - if value is None: - value = arg - arg = None # type: ignore - action_result = f"Unrecognized argument: {value}" - for action in self._pos_actions: - if not action.check_argument(value): - action_result = action.handle(result, value) - break - # check result - if action_result == 'stop': - break - if action_result: - raise ArgumentError(action_result, arg=arg) - # set the rest of the arguments - arguments.reverse() - result.rest += arguments - return result - - def add_argument(self, action_class: Type[Action], *names: str, **kw): - """Add an argument handler - - :param action_class: Action(kw) will be added as a handler - :param names: Option or positional argument name - """ - action = action_class(**kw) - is_opt = False - for name in names: - if not name.startswith('-'): - continue - self._opt_actions[name] = action - is_opt = True - if not is_opt: - if 'metavar' not in kw: - raise ArgumentError(f"Missing metavar for action {action}") - self._pos_actions.append(action) - - def print_help(self): - """Print the help""" - lines = [] - tpl = " {:<27} {}" - if self._opt_actions: - lines.append('options:') - visited = set() - for action in self._opt_actions.values(): - if action in visited: - continue - visited.add(action) - opts = [o for o, a in self._opt_actions.items() if a == action] - option_line = ', '.join(opts) - if action.metavar: - option_line += ' ' + action.metavar - if len(option_line) > 27: - lines.append(tpl.format(option_line, '')) - if action.help: - lines.append((30 * ' ') + action.help) - else: - lines.append(tpl.format(option_line, action.help or '')) - lines.append('') - if self._pos_actions: - lines.append('positional arguments:') - for action in self._pos_actions: - lines.append(tpl.format(action.metavar or '', action.help or '')) - for name, help in self.help_messages.items(): - lines.append(tpl.format(name, help)) - print(*lines, sep='\n') - - -def configure_parser(parser: ArgumentParser, *, app=None): - """Add argument parsing for alphaconf""" - parser.add_argument( - ConfigurationAction, - metavar='key=value', - help='Configuration items', - ) - parser.add_argument( - HelpAction, - '-h', - '--help', - help="Show the help", - ) - if app and app.properties.get('version'): - parser.add_argument( - VersionAction, - '-V', - '--version', - help="Show the version", - ) - parser.add_argument( - ShowConfigurationAction, - '-C', - '--configuration', - help="Show the configuration", - ) - parser.add_argument( - ConfigurationFileAction, - '-f', - '--config', - '--config-file', - metavar='path', - help="Load configuration from file", - ) - parser.add_argument( - ConfigurationSelectAction, - '--select', - help="Shortcut to select a base configuration", - metavar="key=base_template", - ) diff --git a/alphaconf/internal/configuration.py b/alphaconf/internal/configuration.py index b636920..9e895ff 100644 --- a/alphaconf/internal/configuration.py +++ b/alphaconf/internal/configuration.py @@ -1,5 +1,4 @@ import copy -import os import warnings from enum import Enum from typing import ( @@ -75,7 +74,7 @@ def get( default: Union[T, RaiseOnMissingType] = raise_on_missing, ) -> T: ... - def get(self, key: Union[str, Type], type=None, *, default=raise_on_missing): + def get(self, key: Union[str, Type], type=None, *, default: Any = raise_on_missing) -> Any: """Get a configuation value and cast to the correct type""" if isinstance(key, _cla_type): return self.__get_type(key, default=default) @@ -171,29 +170,8 @@ def add_helper(self, key, description): """Assign a helper description""" self.helpers[key] = description - def from_environ(self, prefixes: Iterable[str]) -> DictConfig: - """Load environment variables into a dict configuration""" - from yaml.error import YAMLError # type: ignore - - trans = str.maketrans('_', '.', '"\\=') - prefixes = tuple(prefixes) - dotlist = [ - (name.lower().translate(trans).strip('.'), value) - for name, value in os.environ.items() - if name.startswith(prefixes) - ] - conf = OmegaConf.create({}) - for name, value in dotlist: - name = Configuration._find_name(name.split('.'), self.c) - try: - conf.merge_with_dotlist([f"{name}={value}"]) - except YAMLError: - # if cannot load the value as a dotlist, just add the string - OmegaConf.update(conf, name, value) - return conf - @staticmethod - def _find_name(parts: List[str], conf: DictConfig) -> str: + def _find_name(parts: List[str], conf: DictConfig) -> str: # XXX move to application """Find a name from parts, by trying joining with '.' (default) or '_'""" if len(parts) < 2: return "".join(parts) diff --git a/alphaconf/invoke.py b/alphaconf/invoke.py deleted file mode 100644 index ccf1e73..0000000 --- a/alphaconf/invoke.py +++ /dev/null @@ -1,82 +0,0 @@ -from typing import Dict, Union - -import invoke -from omegaconf import OmegaConf - -from .cli import run as _application_run -from .internal import application, arg_parser - -__doc__ = """Invoke wrapper for an application - -Adding the following lines at the end of the file adds support for alphaconf -to inject the configuration. -Instead of a collection, you could pass `globals()`. - - ns = Collection(...) - alphaconf.invoke.run(__name__, ns) -""" - - -class InvokeAction(arg_parser.Action): - """Apped value to the result and let invoke run (stop parsing)""" - - def handle(self, result, value): - result.rest.append(value) - return 'stop' - - -class InvokeApplication(application.Application): - """Application that launched an invoke.Program""" - - def __init__(self, namespace: invoke.Collection, **properties) -> None: - super().__init__(**properties) - self.namespace = namespace - self.argument_parser.add_argument( - InvokeAction, - metavar="-- invoke arguments", - help="Rest is passed to invoke", - ) - - def _handle_parsed_result(self): - if self.parsed.rest: - return None - return super()._handle_parsed_result() - - def run_program(self): - """Create and run the invoke program""" - argv = [self.name, *self.parsed.rest] - namespace = self.namespace - configuration = OmegaConf.to_object(self.configuration.c) - namespace.configure(configuration) - prog = invoke.Program(namespace=namespace, binary=self.name) - return prog.run(argv) - - -def collection(variables: Dict = {}) -> invoke.Collection: - """Create a new collection base on tasks in the variables""" - return invoke.Collection(*[v for v in variables.values() if isinstance(v, invoke.Task)]) - - -def run( - __name__: str, namespace: Union[invoke.collection.Collection, Dict], **properties -) -> InvokeApplication: - """Create an invoke application and run it if __name__ is __main__""" - if isinstance(namespace, invoke.Collection): - ns = namespace - else: - ns = collection(namespace) - app = InvokeApplication(ns, **properties) - if __name__ == '__main__': - # Let's run the application and parse the arguments - _application_run(app.run_program, app=app) - else: - # Just configure the namespace and set the application - import alphaconf - import alphaconf.logging_util - - alphaconf.set_application(app) - ns.configure(alphaconf.get("")) - alphaconf.logging_util.setup_application_logging( - app.configuration.get('logging', default=None) - ) - return app diff --git a/demo.ipynb b/demo.ipynb index 32b2da6..3269b6b 100644 --- a/demo.ipynb +++ b/demo.ipynb @@ -11,90 +11,21 @@ }, { "cell_type": "code", - "execution_count": 1, + "execution_count": null, "metadata": { "tags": [] }, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "usage: example [arguments] [key=value ...]\n", - "\n", - "Simple demo of alphaconf\n", - "\n", - "options:\n", - " -h, --help Show the help\n", - " -V, --version Show the version\n", - " -C, --configuration Show the configuration\n", - " -f, --config, --config-file path \n", - " Load configuration from file\n", - " --select key=base_template Shortcut to select a base configuration\n", - "\n", - "positional arguments:\n", - " key=value Configuration items\n", - " show The name of the selection to show\n", - " exception If set, raise an exception\n", - "\u001b[0m" - ] - } - ], + "outputs": [], "source": [ "# Show the application help\n", - "!./example-simple.py -h" + "!./example-simple.py --help" ] }, { "cell_type": "code", - "execution_count": 2, + "execution_count": null, "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "logging:\n", - " version: 1\n", - " formatters:\n", - " simple:\n", - " format: '%(asctime)s %(levelname)s %(name)s: %(message)s'\n", - " datefmt: '%H:%M:%S'\n", - " default:\n", - " format: '%(asctime)s %(levelname)s %(name)s [%(process)s,%(threadName)s]: %(message)s'\n", - " color:\n", - " class: alphaconf.logging_util.ColorFormatter\n", - " format: ${..default.format}\n", - " json:\n", - " class: alphaconf.logging_util.JSONFormatter\n", - " handlers:\n", - " console:\n", - " class: logging.StreamHandler\n", - " formatter: color\n", - " stream: ext://sys.stdout\n", - " root:\n", - " handlers:\n", - " - console\n", - " level: WARNING\n", - " disable_existing_loggers: false\n", - "base:\n", - " logging:\n", - " - default\n", - " - none\n", - "exception: false\n", - "server:\n", - " name: test_server\n", - " user: ${oc.env:USER}\n", - "application:\n", - " name: example\n", - " version: '0.1'\n", - "example: config\n", - "arg: name\n", - "\n", - "\u001b[0m" - ] - } - ], + "outputs": [], "source": [ "# Show the configuration, load a file before and set a configuration value\n", "!./example-simple.py -f example-config.yaml arg=name -C" @@ -102,20 +33,9 @@ }, { "cell_type": "code", - "execution_count": 3, + "execution_count": null, "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "app: example\n", - "server.name test_server\n", - "server.user: k\n", - "\u001b[0m" - ] - } - ], + "outputs": [], "source": [ "# Using templates\n", "!./example-simple.py --select logging=none" @@ -123,22 +43,9 @@ }, { "cell_type": "code", - "execution_count": 4, + "execution_count": null, "metadata": {}, - "outputs": [ - { - "name": "stdout", - "output_type": "stream", - "text": [ - "INFO \u001b[32mStart (example-inv: InvokeApplication.run_program)\u001b[0m\n", - "INFO \u001b[32mHello\u001b[0m\n", - "INFO \u001b[32mBackup: me\u001b[0m\n", - "INFO \u001b[32mParam: [4]\u001b[0m\n", - "INFO \u001b[32mEnd.\u001b[0m\n", - "\u001b[0m" - ] - } - ], + "outputs": [], "source": [ "# Invoke integration\n", "!python ./example-inv.py 'logging.formatters.default.format=\"%(levelname)s %(message)s\"' backup=me doit --param 4" @@ -164,7 +71,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.6" + "version": "3.11.8" }, "orig_nbformat": 2 }, diff --git a/example-inv.py b/example-inv.py index cdd40db..9256e63 100644 --- a/example-inv.py +++ b/example-inv.py @@ -2,7 +2,7 @@ from invoke import task -import alphaconf.invoke +import alphaconf @task @@ -16,5 +16,4 @@ def doit(ctx, param=None): # add some default configuration and run/configure invoke's namespace alphaconf.setup_configuration({'backup': 'all'}) -# TODO just setup logging and load variables into ns? -alphaconf.invoke.run(__name__, globals()) +alphaconf.initialize() diff --git a/example-plumbum.py b/example-plumbum.py index cb415c2..97ee7e9 100755 --- a/example-plumbum.py +++ b/example-plumbum.py @@ -4,20 +4,19 @@ import plumbum import alphaconf.cli +from alphaconf.inject import inject_auto alphaconf.setup_configuration({"cmd": "ls"}) -app = alphaconf.cli.CliApplication() - -@app.command() -def main(): +@inject_auto() +def main(cmd: str): """Simple demo of alphaconf with plumbum""" log = logging.getLogger(__name__) - cmd = plumbum.local[alphaconf.get("cmd")] + cmd = plumbum.local[cmd] log.info("Running a command %s", cmd) return cmd.run_fg() if __name__ == '__main__': - app.run() + alphaconf.cli.run(main) diff --git a/example-simple.py b/example-simple.py index 588e964..15ba5dd 100755 --- a/example-simple.py +++ b/example-simple.py @@ -26,10 +26,6 @@ class Opts(BaseModel): ) -app = alphaconf.cli.CliApplication(name='example', version='0.1') - - -@app.command() def main(): """Simple demo of alphaconf""" @@ -62,4 +58,4 @@ def main(): if __name__ == '__main__': - app.run() + alphaconf.cli.run(main, name="example") diff --git a/pyproject.toml b/pyproject.toml index e90049a..1d524d3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,16 +28,11 @@ dependencies = [ [project.optional-dependencies] color = ["colorama"] dotenv = ["python-dotenv"] -invoke = ["invoke"] pydantic = ["pydantic>=2"] -toml = ["toml"] [project.urls] Homepage = "https://github.com/kmagusiak/alphaconf" -[project.scripts] -alphaconf = "alphaconf.cli:run" - [[project.authors]] name = "Krzysztof Magusiak" email = "chrmag@poczta.onet.pl" diff --git a/requirements.txt b/requirements.txt index 62860de..f88c21a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -13,5 +13,8 @@ omegaconf>=2 colorama pydantic>=2 python-dotenv -invoke toml + +# Other tools +invoke +plumbum diff --git a/tests/test_arg_parser.py b/tests/test_arg_parser.py index e148ab3..3466d15 100644 --- a/tests/test_arg_parser.py +++ b/tests/test_arg_parser.py @@ -1,7 +1,7 @@ import pytest from omegaconf import OmegaConf -import alphaconf.internal.arg_parser as ap +import alphaconf.internal.old_arg_parser as ap from alphaconf import Application