Skip to content

Commit

Permalink
Drafting
Browse files Browse the repository at this point in the history
  • Loading branch information
kmagusiak committed May 9, 2024
1 parent 58a4893 commit 2ea76ff
Show file tree
Hide file tree
Showing 16 changed files with 280 additions and 701 deletions.
12 changes: 4 additions & 8 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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/
78 changes: 41 additions & 37 deletions alphaconf/__init__.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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')


#######################################
Expand Down Expand Up @@ -156,4 +160,4 @@ def __alpha_configuration():

# Initialize configuration
__alpha_configuration()
__all__ = ["get", "setup_configuration", "set_application", "Application", "frozendict"]
__all__ = ["get", "setup_configuration", "frozendict"]
151 changes: 85 additions & 66 deletions alphaconf/cli.py
Original file line number Diff line number Diff line change
@@ -1,58 +1,93 @@
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(
main: Callable[[], T],
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.
Expand All @@ -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:
Expand All @@ -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:
Expand Down
2 changes: 1 addition & 1 deletion alphaconf/inject.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 += "."
Expand Down
31 changes: 0 additions & 31 deletions alphaconf/interactive.py

This file was deleted.

Loading

0 comments on commit 2ea76ff

Please sign in to comment.