From 04f6aa7a01571df6e9426265092251b1569afdda Mon Sep 17 00:00:00 2001 From: Krzysztof Magusiak Date: Tue, 26 Sep 2023 20:33:39 +0200 Subject: [PATCH 1/4] Update settings and add fronzenset --- .vscode/settings.json | 5 +---- alphaconf/__init__.py | 12 ++++++----- alphaconf/frozendict.py | 46 ++++++++++++++++++++++++++++++++++++++++ alphaconf/interactive.py | 2 +- 4 files changed, 55 insertions(+), 10 deletions(-) create mode 100644 alphaconf/frozendict.py diff --git a/.vscode/settings.json b/.vscode/settings.json index c806ce1..ac8a2b1 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,12 +1,9 @@ { - "python.formatting.blackArgs": [ + "black-formatter.args": [ "--skip-string-normalization", "--line-length", "100" ], - "python.formatting.provider": "black", - "python.linting.flake8Enabled": true, - "python.linting.mypyEnabled": true, "python.testing.pytestArgs": [ "tests" ], diff --git a/alphaconf/__init__.py b/alphaconf/__init__.py index 84f450f..6f654aa 100644 --- a/alphaconf/__init__.py +++ b/alphaconf/__init__.py @@ -20,6 +20,7 @@ from omegaconf import Container, DictConfig, MissingMandatoryValue, OmegaConf +from .frozendict import frozendict # noqa: F401 (expose) from .internal import Application from .internal.arg_parser import ArgumentError, ExitApplication from .internal.type_resolvers import convert_to_type @@ -271,22 +272,22 @@ def __run_application(app: Application, main: Callable[[], T], exc_info=True) -> set_application(app) app_log = logging.getLogger() if get('testing', bool): - app_log.info('Application testing (%s: %s)', app.name, main.__qualname__) + app_log.info('Testing (%s: %s)', app.name, main.__qualname__) return get('testing') # Run the application try: - app_log.info('Application start (%s: %s)', app.name, main.__qualname__) + app_log.info('Start (%s: %s)', app.name, main.__qualname__) for missing_key in OmegaConf.missing_keys(configuration.get()): app_log.warning('Missing configuration key: %s', missing_key) result = main() if result is None: - app_log.info('Application end.') + app_log.info('End.') else: - app_log.info('Application end: %s', result) + app_log.info('End: %s', result) return result except Exception as e: # no need to log exc_info beacause the parent will handle it - app_log.error('Application failed (%s) %s', type(e).__name__, e, exc_info=exc_info) + app_log.error('Failed (%s) %s', type(e).__name__, e, exc_info=exc_info) raise @@ -303,6 +304,7 @@ def setup_configuration( if isinstance(conf, DictConfig): config = conf else: + # TODO support a.b: v in dicts? created_config = OmegaConf.create(conf) if not (created_config and isinstance(created_config, DictConfig)): raise ValueError('Expecting a non-empty dict configuration') diff --git a/alphaconf/frozendict.py b/alphaconf/frozendict.py new file mode 100644 index 0000000..3b25c3d --- /dev/null +++ b/alphaconf/frozendict.py @@ -0,0 +1,46 @@ +import sys + +# Performance of this small dict is enough for most cases. +# Only when the dict becomes big, the copy strategy is slower after the +# dict contains over a few hundred of elements. So we'll keep it simple, +# hence the primary usage is for small contextual information. + + +class FrozenDict(dict): + """Immutable dict based on a dict() implementation""" + + def _immutable(self, *a, **kw): + raise NotImplementedError('Immutable dict') + + update = _immutable # type: ignore + __setitem__ = _immutable # type: ignore + __delitem__ = _immutable # type: ignore + setdefault = _immutable # type: ignore + update = _immutable # type: ignore + clear = _immutable # type: ignore + pop = _immutable # type: ignore + popitem = _immutable # type: ignore + + def __repr__(self): + return 'frozendict' + super().__repr__() + + def __str__(self): + return super().__repr__() + + @classmethod + def fromkeys(cls, it, v=None): + return FrozenDict((i, v) for i in it) + + if sys.version_info >= (3, 9): + + def __or__(self, value): + return FrozenDict({**self, **value}) + + def __ror__(self, value): + return FrozenDict({**value, **self}) + + __ior__ = _immutable # type: ignore + + +frozendict = FrozenDict +__all__ = ['frozendict'] diff --git a/alphaconf/interactive.py b/alphaconf/interactive.py index 9d8dd44..44e64bf 100644 --- a/alphaconf/interactive.py +++ b/alphaconf/interactive.py @@ -6,7 +6,7 @@ from .internal.load_file import read_configuration_file -__doc__ = """Helpers for interactive applications.""" +__doc__ = """Helpers for interactive applications like ipython.""" __all__ = ['mount', 'read_configuration_file', 'load_configuration_file'] From d898112ac51aea88aca5e21b46bcc8bd7e2ac2b6 Mon Sep 17 00:00:00 2001 From: Krzysztof Magusiak Date: Sun, 26 Nov 2023 23:08:22 +0100 Subject: [PATCH 2/4] New API and pydantic (#21) --- .github/workflows/lint-test.yaml | 12 +- .github/workflows/python-publish.yaml | 4 +- .gitignore | 1 + README.md | 30 +- alphaconf/__init__.py | 301 +++----------------- alphaconf/cli.py | 100 +++++++ alphaconf/frozendict.py | 14 +- alphaconf/interactive.py | 24 +- alphaconf/internal/__init__.py | 387 -------------------------- alphaconf/internal/application.py | 286 +++++++++++++++++++ alphaconf/internal/arg_parser.py | 25 +- alphaconf/internal/configuration.py | 262 +++++++++++++++++ alphaconf/internal/dotenv_vars.py | 31 +++ alphaconf/internal/type_resolvers.py | 12 + alphaconf/invoke.py | 31 ++- alphaconf/logging_util.py | 24 +- demo.ipynb | 20 +- example-inv.py | 10 +- example-simple.py | 42 +-- example-typed.py | 31 ++- pre-commit | 10 +- pyproject.toml | 5 +- requirements.txt | 1 + tests/conftest.py | 13 +- tests/test_alphaconf.py | 76 ++--- tests/test_arg_parser.py | 2 +- tests/test_configuration.py | 103 +++++++ tests/test_configuration_typed.py | 100 +++++++ tests/test_logging.py | 10 +- tests/test_select_and_type.py | 90 ------ 30 files changed, 1160 insertions(+), 897 deletions(-) create mode 100644 alphaconf/cli.py create mode 100644 alphaconf/internal/application.py create mode 100644 alphaconf/internal/configuration.py create mode 100644 alphaconf/internal/dotenv_vars.py create mode 100644 tests/test_configuration.py create mode 100644 tests/test_configuration_typed.py delete mode 100644 tests/test_select_and_type.py diff --git a/.github/workflows/lint-test.yaml b/.github/workflows/lint-test.yaml index 73a8b64..233d5e3 100644 --- a/.github/workflows/lint-test.yaml +++ b/.github/workflows/lint-test.yaml @@ -10,12 +10,12 @@ jobs: lint: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v3 - name: Set up Python - uses: actions/setup-python@v2 + uses: actions/setup-python@v4 with: # use minimum version here from setup.py - python-version: "3.8" + python-version: "3.9" - name: Install dependencies run: | python -m pip install --upgrade pip @@ -28,11 +28,11 @@ jobs: strategy: matrix: # lowest, common (defaut ubuntu LTS), newest - python-version: ["3.8", "3.10", "3.11"] + python-version: ["3.9", "3.10", "3.11"] steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v2 + uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Install dependencies diff --git a/.github/workflows/python-publish.yaml b/.github/workflows/python-publish.yaml index fdb899b..fdcd2b6 100644 --- a/.github/workflows/python-publish.yaml +++ b/.github/workflows/python-publish.yaml @@ -21,12 +21,12 @@ jobs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v2 + - uses: actions/checkout@v3 with: # don't use shallow checkout to determine an intermediary version correctly fetch-depth: 0 - name: Set up Python - uses: actions/setup-python@v2 + uses: actions/setup-python@v4 with: python-version: '3.11' - name: Install dependencies diff --git a/.gitignore b/.gitignore index 57c9aba..76075c1 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,7 @@ __pycache__ # Local env .env .venv +/local* # Other tools dist diff --git a/README.md b/README.md index d02ac27..2cb9fc1 100644 --- a/README.md +++ b/README.md @@ -5,8 +5,8 @@ A small library to ease writing parameterized scripts. The goal is to execute a single script and be able to overwrite the parameters easily. -The configuration is based on [OmegaConf](https://omegaconf.readthedocs.io/). -Optionally, loading from toml is possible. +The configuration is based on [OmegaConf]. +Optionally, loading from toml or using [pydantic] is possible. To run multiple related tasks, there is an integration with [invoke](https://www.pyinvoke.org). @@ -23,10 +23,9 @@ To run an application, you need... import alphaconf import logging # define the default values and helpers -alphaconf.setup_configuration(""" -server: - url: http://default -""", { +alphaconf.setup_configuration({ + "server.url": "http://default", +}, { "server.url": "The URL to show here", }) @@ -36,7 +35,7 @@ def main(): log.info('has server.user:', alphaconf.get('server.user', bool)) if __name__ == '__main__': - alphaconf.run(main) + alphaconf.cli.run(main) ``` Invoking: @@ -44,7 +43,7 @@ Invoking: python myapp.py server.url=http://github.com ``` -During an interactive session, you can set the application in the current +During an *interactive session*, you can set the application in the current context. ```python # import other modules @@ -64,7 +63,7 @@ Then configuration is built from: - default configurations defined using (`alphaconf.setup_configuration`) - `application` key is generated -- PYTHON_ALPHACONF may contain a path to a configuration file +- `PYTHON_ALPHACONF` environment variable may contain a path to load - configuration files from configuration directories (using application name) - environment variables based on key prefixes, except "BASE" and "PYTHON"; @@ -77,13 +76,14 @@ Finally, the configuration is fully resolved and logging is configured. ## Configuration templates and resolvers -Omegaconf's resolvers may be used as configuration values. +[OmegaConf]'s resolvers may be used as configuration values. For example, `${oc.env:USER,me}` would resolve to the environment variable USER with a default value "me". Similarly, `${oc.select:path}` will resolve to another configuration value. Additional resolvers are added to read file contents. These are the same as type casts: read_text, read_strip, read_bytes. +-- TODO use secrets for v1 The select is used to build multiple templates for configurations by providing base configurations. @@ -96,6 +96,7 @@ dict defined in base.logging.default and you can select it using ## Configuration values and integrations ### Typed-configuration +-- TODO update to pydantic You can use *omegaconf* with *dataclasses* to specify which values are enforced in the configuration. @@ -123,7 +124,10 @@ alphaconf.setup_configuration({'backup': 'all'}) alphaconf.invoke.run(__name__, ns) ``` -### Interactive and manual usage +## Way to 1.0 +- Secret management +- Install completions for bash +- Run a function after importing the module -Use `alphaconf.interactive.mount()` or load manually create an -`alphaconf.Application`, configure it and set it. +[OmegaConf]: https://omegaconf.readthedocs.io/ +[pydantic]: https://docs.pydantic.dev/latest/ diff --git a/alphaconf/__init__.py b/alphaconf/__init__.py index 6f654aa..fcffc63 100644 --- a/alphaconf/__init__.py +++ b/alphaconf/__init__.py @@ -1,29 +1,10 @@ -import contextlib -import contextvars -import logging import re -import sys -from contextvars import ContextVar -from typing import ( - Any, - Callable, - Dict, - List, - Literal, - Optional, - Type, - TypeVar, - Union, - cast, - overload, -) - -from omegaconf import Container, DictConfig, MissingMandatoryValue, OmegaConf +import warnings +from typing import Callable, Optional, Sequence, TypeVar, Union from .frozendict import frozendict # noqa: F401 (expose) -from .internal import Application -from .internal.arg_parser import ArgumentError, ExitApplication -from .internal.type_resolvers import convert_to_type +from .internal.application import Application +from .internal.configuration import Configuration __doc__ = """AlphaConf @@ -41,173 +22,56 @@ configuration load and logging setup. if __name__ == '__main__': - alphaconf.run(main) + alphaconf.cli.run(main) """ -"""A list of functions which given a key indicate whether it's a secret""" SECRET_MASKS = [ # mask if contains a kind of secret and it's not in a file re.compile(r'.*(key|password|secret)s?(?!_file)(_|$)|^private(_key|$)').match, ] - -####################################### -# APPLICATION CONTEXT - -"""The current configuration""" -configuration: ContextVar[DictConfig] = ContextVar('configuration', default=OmegaConf.create()) -"""Additional helpers for the application""" -_helpers: ContextVar[Dict[str, str]] = ContextVar('configuration_helpers', default={}) +"""A list of functions which given a key indicate whether it's a secret""" T = TypeVar('T') +####################################### +# APPLICATION CONTEXT +# ContextVar are no more used because some executions frameworks reset +# the context. -@overload -def select( - container: Any, - key: str, - type: Type[T], - *, - default: Optional[T] = None, - required: Literal[True], -) -> T: - ... - - -@overload -def select( - container: Any, - key: str, - type: Type[T], - *, - default: Optional[T] = None, - required: bool = False, -) -> Optional[T]: - ... - - -@overload -def select( - container: Any, - key: str, - type: Union[str, Type[T], None] = None, - *, - default: Any = None, - required: bool = False, -) -> Any: - ... - - -def select(container: Any, key: str, type=None, *, default=None, required: bool = False) -> Any: - """Select a configuration item from the container - - :param container: The container to select from (Container, dict, etc.) - :param key: The selection key - :param type: The type of the object to return - :param default: The default value is selected value is None - :param required: Raise MissingMandatoryValue if the selected value and default are None - :return: The selected value in the container - """ - c: Any - # make sure we have a container and select from it - if isinstance(container, Container): - c = container - else: - c = OmegaConf.create(container) - c = OmegaConf.select(c, key, throw_on_missing=required) - # handle empty result - if c is None: - if default is None and required: - raise MissingMandatoryValue("Key not found: %s" % key) - return default - # check the returned type and convert when necessary - if type is not None and isinstance(c, type): - return c - if isinstance(c, Container): - c = OmegaConf.to_object(c) - if type is not None: - c = convert_to_type(c, type) - return c - - -@overload -def get( - config_key: str, - type: Type[T], - *, - default: Optional[T] = None, - required: Literal[True], -) -> T: - ... - - -@overload -def get( - config_key: str, - type: Type[T], - *, - default: Optional[T] = None, - required: bool = False, -) -> Optional[T]: - ... - - -@overload -def get( - config_key: str, - type: Union[str, Type[T], None] = None, - *, - default: Any = None, - required: bool = False, -) -> Any: - ... +_global_configuration: Configuration = Configuration() +"""The global configuration""" +setup_configuration = _global_configuration.setup_configuration -def get(config_key: str, type=None, *, default=None, required: bool = False) -> Any: - """Select a configuration item from the current configuration""" - return select(configuration.get(), config_key, type=type, default=default, required=required) +_application: Optional[Application] = None +get = _global_configuration.get -@contextlib.contextmanager -def set(**kw): - """Update the configuration in a with block +def set_application(app: Application) -> None: + """Setup the application globally - with alphaconf.set(a=value): - assert alphaconf.get('a') == value + This loads the configuration and initializes the application. + The function may raise ExitApplication. """ - if not kw: - yield + global _application, get + if _application is app: return - config = configuration.get() - # merging 2 dict-like objects - config = cast(DictConfig, OmegaConf.merge(config, kw)) - token = configuration.set(config) - yield - configuration.reset(token) - - -def set_application(app: Application, merge: bool = False): - """Set the application and its configuration - - :param app: The application - :param merge: Wether to merge the current configuration with the application (default false) - """ - config = app.configuration - if merge: - # merging 2 DictConfig objects - config = cast(DictConfig, OmegaConf.merge(configuration.get(), config)) - configuration.set(config) + if _application is not None: + _application.log.info("Another application will be loaded") + _application = app + get = app.configuration.get def run( main: Callable[[], T], - arguments: Union[bool, List[str]] = True, + arguments: Union[bool, Sequence[str]] = True, *, should_exit: bool = True, app: Optional[Application] = None, **config, ) -> Optional[T]: - """Run this application + """Run this application (deprecated) 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. @@ -218,106 +82,10 @@ def run( :param config: Arguments passed to Application.__init__() and Application.setup_configuration() :return: The result of main """ - from .internal import application_log as log - - if 'setup_logging' not in config: - config['setup_logging'] = True - 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) - try: - app.setup_configuration(arguments, **config) - except MissingMandatoryValue as e: - log.error(e) - if should_exit: - sys.exit(99) - raise - except ArgumentError as e: - log.error(e) - if should_exit: - sys.exit(2) - raise - except ExitApplication: - log.debug('Normal application exit') - if should_exit: - sys.exit() - return None - context = contextvars.copy_context() - try: - return context.run(__run_application, app=app, main=main, exc_info=should_exit) - except Exception: - if should_exit: - log.debug('Exit application') - sys.exit(1) - raise + warnings.warn("use alphaconf.cli.run directly", DeprecationWarning) + from .cli import run - -def __run_application(app: Application, main: Callable[[], T], exc_info=True) -> T: - """Set the application and execute main""" - set_application(app) - app_log = logging.getLogger() - if get('testing', bool): - app_log.info('Testing (%s: %s)', app.name, main.__qualname__) - return get('testing') - # Run the application - try: - app_log.info('Start (%s: %s)', app.name, main.__qualname__) - for missing_key in OmegaConf.missing_keys(configuration.get()): - app_log.warning('Missing configuration key: %s', missing_key) - result = main() - if result is None: - app_log.info('End.') - else: - app_log.info('End: %s', result) - return result - except Exception as e: - # no need to log exc_info beacause the parent will handle it - app_log.error('Failed (%s) %s', type(e).__name__, e, exc_info=exc_info) - raise - - -def setup_configuration( - conf: Union[DictConfig, str, Dict], - helpers: Dict[str, str] = {}, -): - """Add a default configuration - - :param conf: The configuration to merge into the global configuration - :param helpers: Description of parameters used in argument parser helpers - """ - # merge the configurations - if isinstance(conf, DictConfig): - config = conf - else: - # TODO support a.b: v in dicts? - created_config = OmegaConf.create(conf) - if not (created_config and isinstance(created_config, DictConfig)): - raise ValueError('Expecting a non-empty dict configuration') - config = created_config - # merging 2 DictConfig - config = cast(DictConfig, OmegaConf.merge(configuration.get(), config)) - configuration.set(config) - # setup helpers - for h_key in helpers: - key = h_key.split('.', 1)[0] - if not config or key not in config: - raise ValueError('Invalid helper not in configuration [%s]' % key) - _helpers.set({**_helpers.get(), **helpers}) + return run(main, arguments, should_exit=should_exit, app=app, **config) ####################################### @@ -367,9 +135,13 @@ def __alpha_configuration(): ' %(name)s [%(process)s,%(threadName)s]: %(message)s', }, }, - 'handlers': {}, + 'handlers': { + 'null': { + 'class': 'logging.NullHandler', + }, + }, 'root': { - 'handlers': [], + 'handlers': ['null'], 'level': 'INFO', }, } @@ -382,3 +154,4 @@ def __alpha_configuration(): # Initialize configuration __alpha_configuration() +__all__ = ["get", "setup_configuration", "set_application", "Application", "frozendict"] diff --git a/alphaconf/cli.py b/alphaconf/cli.py new file mode 100644 index 0000000..8aa674c --- /dev/null +++ b/alphaconf/cli.py @@ -0,0 +1,100 @@ +import sys +from typing import Callable, Optional, Sequence, TypeVar, Union + +from omegaconf import MissingMandatoryValue, OmegaConf + +from . import set_application +from .internal.application import Application +from .internal.arg_parser import ArgumentError, ExitApplication + +T = TypeVar('T') + + +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 + + 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. + + :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 + """ + # 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 + try: + 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 + if setup_logging: + from .logging_util import setup_application_logging + + setup_application_logging(configuration.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: + sys.exit(2) + raise + except ExitApplication: + log.debug('Normal application exit') + 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__) + return None + try: + log.info('Start (%s: %s)', app.name, main.__qualname__) + for missing_key in OmegaConf.missing_keys(configuration.c): + log.warning('Missing configuration key: %s', missing_key) + result = main() + if result is None: + log.info('End.') + else: + log.info('End: %s', result) + return result + except Exception as e: + # no need to log exc_info beacause the parent will handle it + log.error('Failed (%s) %s', type(e).__name__, e, exc_info=should_exit) + raise diff --git a/alphaconf/frozendict.py b/alphaconf/frozendict.py index 3b25c3d..b06e7bb 100644 --- a/alphaconf/frozendict.py +++ b/alphaconf/frozendict.py @@ -1,5 +1,3 @@ -import sys - # Performance of this small dict is enough for most cases. # Only when the dict becomes big, the copy strategy is slower after the # dict contains over a few hundred of elements. So we'll keep it simple, @@ -31,15 +29,13 @@ def __str__(self): def fromkeys(cls, it, v=None): return FrozenDict((i, v) for i in it) - if sys.version_info >= (3, 9): - - def __or__(self, value): - return FrozenDict({**self, **value}) + def __or__(self, value): + return FrozenDict({**self, **value}) - def __ror__(self, value): - return FrozenDict({**value, **self}) + def __ror__(self, value): + return FrozenDict({**value, **self}) - __ior__ = _immutable # type: ignore + __ior__ = _immutable # type: ignore frozendict = FrozenDict diff --git a/alphaconf/interactive.py b/alphaconf/interactive.py index 44e64bf..004b654 100644 --- a/alphaconf/interactive.py +++ b/alphaconf/interactive.py @@ -1,22 +1,26 @@ import logging from typing import List -# absolute import to make sure it's properly initialized -import alphaconf - +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 a new application with a setup configuration""" - app = alphaconf.Application(name='interactive') - app.setup_configuration( - arguments=False, configuration_paths=configuration_paths, setup_logging=setup_logging - ) - alphaconf.set_application(app, merge=True) + """Mount the interactive application and setup configuration""" + application.setup_configuration(configuration_paths=configuration_paths) + set_application(application) + if setup_logging: + import logging_util + + logging_util.setup_application_logging( + application.configuration.get('logging'), default=None + ) logging.info('Mounted interactive application') @@ -24,4 +28,4 @@ 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) - alphaconf.setup_configuration(config) + application.configuration.setup_configuration(config) diff --git a/alphaconf/internal/__init__.py b/alphaconf/internal/__init__.py index 8e743a1..e69de29 100644 --- a/alphaconf/internal/__init__.py +++ b/alphaconf/internal/__init__.py @@ -1,387 +0,0 @@ -import itertools -import logging -import os -import sys -import uuid -from typing import Any, Dict, Iterable, List, Optional, Tuple, Union, cast - -from omegaconf import DictConfig, OmegaConf - -from . import arg_parser, load_file - -__doc__ = """Representation of an application with its configuration.""" - -application_log = logging.getLogger(__name__) - - -class Application: - """An application description - - :param properties: Properties of the application, such as: - name, version, short_description, description, etc. - """ - - __config: Optional[DictConfig] - __name: str - version: Optional[str] - description: Optional[str] - short_description: Optional[str] - parsed: Optional[arg_parser.ParseResult] - argument_parser: arg_parser.ArgumentParser - - def __init__( - self, *, name=None, version=None, description=None, short_description=None - ) -> None: - """Initialize the application. - - Properties: - - name: the name of the application (always updated) - - version: version number - - description: description shown in help - - short_description: shorter description - - :param properties: Properties for the app - """ - self.__config = None # initialize - self.__name = name or self.__get_default_name() - self.version = version - self.description = description - self.short_description = short_description - # Add argument parser - self.__initialize_parser() - - def __initialize_parser(self): - from .. import _helpers - - self.parsed = None - self.argument_parser = parser = arg_parser.ArgumentParser() - arg_parser.configure_parser(parser, app=self) - parser.help_messages.update(_helpers.get()) - - @staticmethod - def __get_default_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 _app_configuration(self) -> DictConfig: - """Get the application configuration key""" - return OmegaConf.create( - { - 'application': { - 'name': self.name, - 'version': self.version or '', - 'uuid': str(uuid.uuid4()), - }, - } - ) - - @property - def name(self) -> str: - """Get the name of the application""" - return self.__name - - @property - def configuration(self) -> DictConfig: - """Get the configuration of the application, initialize if necessary""" - if self.__config is None: - self.setup_configuration( - arguments=False, resolve_configuration=False, setup_logging=False - ) - application_log.info('alphaconf initialized') - assert self.__config is not None - return self.__config - - def _get_possible_configuration_paths(self) -> Iterable[str]: - """List of paths where to find configuration files""" - name = self.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 = os.path.expandvars(path) - if path and '$' not in path: - for ext in load_file.SUPPORTED_EXTENSIONS: - yield path.format(name + '.' + ext) - - def _load_dotenv(self, load_dotenv: Optional[bool] = None): - """Load dotenv variables (optionally)""" - if load_dotenv is False: - return - try: - import dotenv - except ModuleNotFoundError: - if load_dotenv: - raise - application_log.debug('dotenv is not installed') - return - path = dotenv.find_dotenv(usecwd=True) - application_log.debug('Loading dotenv: %s', path or '(none)') - if not path: - return - dotenv.load_dotenv(path) - # check local overrides - path += '.local' - if os.path.isfile(path): - dotenv.load_dotenv(path) - - def __load_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), value) - for name, value in os.environ.items() - if name.startswith(prefixes) - ] - conf = OmegaConf.create({}) - for name, value in dotlist: - try: - conf.merge_with_dotlist(["%s=%s" % (name, value)]) - except YAMLError: - # if cannot load the value as a dotlist, just add the string - OmegaConf.update(conf, name, value) - return conf - - def _get_configurations( - self, - configuration_paths: List[str] = [], - env_prefixes: Union[bool, Iterable[str]] = True, - ) -> Iterable[DictConfig]: - """List of all configurations that can be loaded automatically - - - All of the default configurations - - 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) - """ - from .. import configuration as ctx_configuration - - application_log.debug('Loading default and app configurations') - default_configuration = ctx_configuration.get() - yield default_configuration - yield self._app_configuration() - # Read files - env_configuration_path = os.environ.get('PYTHON_ALPHACONF') or '' - for path in itertools.chain( - [env_configuration_path], - self._get_possible_configuration_paths(), - configuration_paths, - ): - if not os.path.isfile(path): - continue - application_log.debug('Load configuration from %s', path) - yield load_file.read_configuration_file(path) - # Environment - prefixes: Optional[Tuple[str, ...]] - if env_prefixes is True: - application_log.debug('Detecting accepted env prefixes') - default_keys = {str(k) for k in default_configuration.keys()} - 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: - application_log.debug('Loading env configuration from prefixes %s' % (prefixes,)) - yield self.__load_environ(prefixes) - - def setup_configuration( - self, - arguments: Union[bool, List[str]] = True, - *, - load_dotenv: Optional[bool] = None, - env_prefixes: Union[bool, Iterable[str]] = True, - configuration_paths: List[str] = [], - resolve_configuration: bool = True, - setup_logging: bool = False, - ) -> None: - """Setup the application configuration - - Can be called only once to setup the configuration and initialize the application. - The function may raise ExitApplication. - - :param arguments: The argument list to parse (default: True to parse sys.argv) - :param load_dotenv: Whether to load dotenv environment (default: yes if installed) - :param env_prefixes: The env prefixes to load the configuration values from (default: auto) - :param resolve_configuration: Test whether the configuration can be resolved (default: True) - :param setup_logging: Whether to setup logging (default: True) - """ - if self.__config is not None: - raise RuntimeError('Configuration already set') - application_log.debug('Start setup application') - - # Parse arguments - if arguments is True: - arguments = sys.argv[1:] - if not isinstance(arguments, list): - arguments = [] - self.parsed = self.argument_parser.parse_args(arguments) - - # Load and merge configurations - self._load_dotenv(load_dotenv=load_dotenv) - configurations = list( - self._get_configurations( - env_prefixes=env_prefixes, - configuration_paths=configuration_paths, - ) - ) - if self.parsed: - configurations.extend(self.parsed.configurations()) - self.__config = cast(DictConfig, OmegaConf.merge(*configurations)) - application_log.debug('Merged %d configurations', len(configurations)) - - # Handle the result - self._handle_parsed_result() - - # Try to get the whole configuration to resolve links - if resolve_configuration: - OmegaConf.resolve(self.__config) - - # Logging - if setup_logging: - application_log.debug('Setup logging') - self._setup_logging() - - def _handle_parsed_result(self): - """Handle result that is in self.parsed""" - 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}") - return None - - def _setup_logging(self) -> None: - """Setup logging - - Set the time to GMT, log key 'logging' from configuration or if none, base logging. - """ - import logging - - from .. import logging_util - - logging_util.set_gmt() - log = logging.getLogger() - logging_config = cast(Dict[str, Any], OmegaConf.to_object(self.configuration.logging)) - if logging_config: - # Configure using the st configuration - import logging.config - - logging.config.dictConfig(logging_config) - elif len(log.handlers) == 0: - # Default logging if not yet initialized - output = logging.StreamHandler() - output.setFormatter(logging_util.ColorFormatter()) - log.addHandler(output) - log.setLevel(logging.INFO) - - def masked_configuration( - self, - *, - mask_base: bool = True, - mask_secrets: bool = True, - mask_keys: List[str] = ['application.uuid'], - ) -> dict: - """Get the configuration as dict with masked items - - :param mask_base: Whether to mask "base" entry - :param mask_secrets: Whether to mask secret keys - :param mask_keys: Which keys to mask - :return: Configuration copy with masked values - """ - from .. import SECRET_MASKS - - config = cast(dict, OmegaConf.to_container(self.configuration)) - if mask_secrets: - config = Application.__mask_config( - config, - lambda p: any(mask(p) for mask in SECRET_MASKS), - lambda v: v if v is None or v == '???' else '*****', - ) - if mask_base and 'base' not in mask_keys: - # first remove all values if the object is not resolved - config['base'] = Application.__mask_config( - config['base'], - lambda p: not isinstance( - OmegaConf.select(self.configuration, p, throw_on_resolution_failure=False), - DictConfig, - ), - lambda v: {}, - ) - # then collapse dict[str,None] into a list[str] - config['base'] = Application.__mask_config( - config['base'], - lambda _: True, - lambda v: list(v) if isinstance(v, dict) and not any(v.values()) else v, - ) - if mask_keys: - config = Application.__mask_config(config, lambda p: p in mask_keys, lambda _: None) - return config - - @staticmethod - def __mask_config(obj, check, replace, path=''): - """Alter the configuration dict - - :param config: The value to mask - :param check: Function to check if we replace the value (path: str) -> bool - :param replace: Function doing the replacement of the value (v) -> Any - :param key: Current path - :return: The modified config - """ - if check(path): - obj = replace(obj) - if isinstance(obj, dict): - result = {} - for key, value in obj.items(): - new = Application.__mask_config( - value, check, replace, f"{path}.{key}" if path else key - ) - if new is not None: - result[key] = new - obj = result - elif isinstance(obj, list): - obj = [ - Application.__mask_config(v, check, replace, f"{path}[{i}]") - for i, v in enumerate(obj) - ] - return obj - - def print_help(self, *, usage=None, description=None, arguments=True): - """Print the help message - Set the arguments to False to disable printing them.""" - if usage is None: - usage = f"usage: {self.name or 'app'}" - usage += " [arguments] [key=value ...]" - if isinstance(usage, str): - print(usage) - if description is None: - description = self.description - if isinstance(description, str): - print() - print(description) - if arguments: - print() - self.argument_parser.print_help() - - def __str__(self) -> str: - ready = self.__config is not None - return f"{type(self).__name__}({self.name}; loaded={ready})" diff --git a/alphaconf/internal/application.py b/alphaconf/internal/application.py new file mode 100644 index 0000000..2815921 --- /dev/null +++ b/alphaconf/internal/application.py @@ -0,0 +1,286 @@ +import itertools +import logging +import os +import sys +import uuid +from typing import Iterable, List, MutableMapping, Optional, Tuple, Union, cast + +from omegaconf import DictConfig, OmegaConf + +from . import arg_parser, load_file +from .configuration import Configuration + + +class Application: + """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 + + def __init__( + self, + *, + name=None, + **properties, + ) -> None: + """Initialize the application. + + Properties: + - name: the name of the application (always updated) + - version + - description + - ... + """ + self.__config = None # initialize + self.__name = name or self.__get_default_name() + self.properties = properties + self.argument_parser = self._build_argument_parser() + + def _build_argument_parser(self) -> arg_parser.ArgumentParser: + from .. import _global_configuration + + p = arg_parser.ArgumentParser(_global_configuration.helpers) + arg_parser.configure_parser(p, app=self) + return p + + @staticmethod + def __get_default_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 _app_configuration(self) -> DictConfig: + """Get the application configuration key""" + return OmegaConf.create( + { + 'application': { + 'name': self.name, + 'version': self.properties.get('version') or '', + 'uuid': str(uuid.uuid4()), + }, + } + ) + + @property + def name(self) -> str: + """Get the name of the application""" + return self.__name + + @property + def configuration(self) -> Configuration: + """Get the configuration of the application, initialize if necessary""" + if self.__config is None: + self.setup_configuration() + self.log.info('alphaconf initialized') + assert self.__config is not None + return self.__config + + def _get_possible_configuration_paths(self) -> Iterable[str]: + """List of paths where to find configuration files""" + name = self.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( + self, + 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) + """ + self.log.debug('Loading default and app configurations') + assert self.__config is not None + default_configuration = self.__config.c + yield default_configuration + yield self._app_configuration() + # Read files + env_configuration_path = os.environ.get('PYTHON_ALPHACONF') or '' + for path in itertools.chain( + [env_configuration_path], + self._get_possible_configuration_paths(), + configuration_paths, + ): + if not os.path.isfile(path): + continue + self.log.debug('Load configuration from %s', path) + yield load_file.read_configuration_file(path) + # Environment + prefixes: Optional[Tuple[str, ...]] + if env_prefixes is True: + self.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: + self.log.debug('Loading env configuration from prefixes %s', prefixes) + yield self.__config.from_environ(prefixes) + if self.parsed: + yield from self.parsed.configurations() + + def setup_configuration( + self, + *, + arguments: List[str] = [], + configuration_paths: Iterable[str] = [], + load_dotenv: Optional[bool] = None, + env_prefixes: Union[bool, Iterable[str]] = True, + resolve_configuration: bool = True, + ): + from .. import _global_configuration as ctx_configuration + from .dotenv_vars import try_dotenv + + try_dotenv(load_dotenv=load_dotenv) + + self.log.debug('Parse arguments') + self.parsed = self.argument_parser.parse_args(arguments) + + self.log.debug('Start setup configuration') + self.__config = Configuration(parent=ctx_configuration) + self.__config._merge( + self._get_configurations( + configuration_paths=configuration_paths, env_prefixes=env_prefixes + ) + ) + self.log.debug('Merged configurations') + + # Handle the result + self._handle_parsed_result() + + # Try to get the whole configuration to resolve links + if resolve_configuration: + OmegaConf.resolve(self.__config.c) + + def _handle_parsed_result(self): + """Handle result that is in self.parsed""" + 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}") + return None + + def masked_configuration( + self, + *, + mask_base: bool = True, + mask_secrets: bool = True, + mask_keys: List[str] = ['application.uuid'], + ) -> dict: + """Get the configuration as dict with masked items + + :param mask_base: Whether to mask "base" entry + :param mask_secrets: Whether to mask secret keys + :param mask_keys: Which keys to mask + :return: Configuration copy with masked values + """ + from .. import SECRET_MASKS + + config = cast(dict, OmegaConf.to_container(self.configuration.c)) + if mask_secrets: + config = Application.__mask_config( + config, + lambda p: any(mask(p) for mask in SECRET_MASKS), + lambda v: v if v is None or v == '???' else '*****', + ) + if mask_base and 'base' not in mask_keys: + # first remove all values if the object is not resolved + config['base'] = Application.__mask_config( + config['base'], + lambda p: not isinstance( + OmegaConf.select(self.configuration.c, p, throw_on_resolution_failure=False), + DictConfig, + ), + lambda v: {}, + ) + # then collapse dict[str,None] into a list[str] + config['base'] = Application.__mask_config( + config['base'], + lambda _: True, + lambda v: list(v) if isinstance(v, dict) and not any(v.values()) else v, + ) + if mask_keys: + config = Application.__mask_config(config, lambda p: p in mask_keys, lambda _: None) + return config + + @staticmethod + def __mask_config(obj, check, replace, path=''): + """Alter the configuration dict + + :param config: The value to mask + :param check: Function to check if we replace the value (path: str) -> bool + :param replace: Function doing the replacement of the value (v) -> Any + :param key: Current path + :return: The modified config + """ + if check(path): + obj = replace(obj) + if isinstance(obj, dict): + result = {} + for key, value in obj.items(): + new = Application.__mask_config( + value, check, replace, f"{path}.{key}" if path else key + ) + if new is not None: + result[key] = new + obj = result + elif isinstance(obj, list): + obj = [ + Application.__mask_config(v, check, replace, f"{path}[{i}]") + for i, v in enumerate(obj) + ] + return obj + + def print_help(self, *, arguments=True): + """Print the help message + Set the arguments to False to disable printing them.""" + prop = self.properties + if usage := prop.get('usage'): + print(usage) + else: + usage = f"usage: {self.name or 'app'}" + usage += " [arguments] [key=value ...]" + print(usage) + if description := (prop.get('description') or prop.get('short_description')): + print() + print(description) + if arguments: + print() + self.argument_parser.print_help() + + def __str__(self) -> str: + ready = self.__config is not None + return f"{type(self).__name__}({self.name}; loaded={ready})" diff --git a/alphaconf/internal/arg_parser.py b/alphaconf/internal/arg_parser.py index 17464b9..c140248 100644 --- a/alphaconf/internal/arg_parser.py +++ b/alphaconf/internal/arg_parser.py @@ -1,5 +1,5 @@ import itertools -from typing import Dict, Iterable, List, Optional, Tuple, Type, Union, cast +from typing import Dict, Iterable, List, Mapping, Optional, Tuple, Type, Union, cast from omegaconf import DictConfig, OmegaConf @@ -74,9 +74,13 @@ class VersionAction(Action): def run(self, app): prog = app.name - version = app.version - print(f"{prog} {version}") - desc = app.short_description + 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 @@ -167,12 +171,12 @@ class ArgumentParser: _opt_actions: Dict[str, Action] _pos_actions: List[Action] - help_messages: Dict[str, str] + help_messages: Mapping[str, str] - def __init__(self) -> None: + def __init__(self, help_messages: Mapping[str, str] = {}) -> None: self._opt_actions = {} self._pos_actions = [] - self.help_messages = {} + self.help_messages = help_messages or {} def parse_args(self, arguments: List[str]) -> ParseResult: """Parse the argument""" @@ -269,9 +273,8 @@ def print_help(self): lines.append('positional arguments:') for action in self._pos_actions: lines.append(tpl.format(action.metavar or '', action.help or '')) - if self.help_messages: - for name, help in self.help_messages.items(): - lines.append(tpl.format(name, help)) + for name, help in self.help_messages.items(): + lines.append(tpl.format(name, help)) print(*lines, sep='\n') @@ -288,7 +291,7 @@ def configure_parser(parser: ArgumentParser, *, app=None): '--help', help="Show the help", ) - if app and app.version: + if app and app.properties.get('version'): parser.add_argument( VersionAction, '-V', diff --git a/alphaconf/internal/configuration.py b/alphaconf/internal/configuration.py new file mode 100644 index 0000000..4f958f1 --- /dev/null +++ b/alphaconf/internal/configuration.py @@ -0,0 +1,262 @@ +import copy +import os +import typing +import warnings +from enum import Enum +from typing import ( + Any, + Dict, + Iterable, + MutableMapping, + Optional, + Type, + TypeVar, + Union, + cast, + overload, +) + +from omegaconf import Container, DictConfig, OmegaConf + +from .type_resolvers import convert_to_type, pydantic + +T = TypeVar('T') + + +class RaiseOnMissingType(Enum): + RAISE = 'raise' + + +raise_on_missing = RaiseOnMissingType.RAISE +_cla_type = type + + +class Configuration: + c: DictConfig + __type_path: MutableMapping[Type, Optional[str]] + __type_value: MutableMapping[Type, Any] + helpers: Dict[str, str] + + def __init__(self, *, parent: Optional["Configuration"] = None) -> None: + if parent: + self.c = OmegaConf.create(parent.c) + self.helpers = copy.copy(parent.helpers) + self.__type_path = copy.copy(parent.__type_path) + else: + self.c = OmegaConf.create({}) + self.helpers = {} + self.__type_path = {} + self.__type_value = {} + + @overload + def get( + self, + key: str, + type: Type[T], + *, + default: Union[T, RaiseOnMissingType] = raise_on_missing, + ) -> T: + ... + + @overload + def get( + self, + key: str, + type: Union[str, Type[T], None] = None, + *, + default: Any = raise_on_missing, + ) -> Any: + ... + + @overload + def get( + self, + key: Type[T], + type: None = None, + *, + default: Union[T, RaiseOnMissingType] = raise_on_missing, + ) -> T: + ... + + def get(self, key: Union[str, Type], type=None, *, default=raise_on_missing): + """Get a configuation value and cast to the correct type""" + if isinstance(key, _cla_type): + return self.__get_type(key, default=default) + # get using a string key + assert isinstance(key, str), "Expecting a str key" + value = OmegaConf.select( + self.c, + key, + default=raise_on_missing, + ) + if value is raise_on_missing: + if default is raise_on_missing: + raise ValueError(f"No value for: {key}") + return default + # check the returned type and convert when necessary + if type is not None and isinstance(value, type): + return value + if isinstance(value, Container): + value = OmegaConf.to_object(value) + if type is not None and default is not None: + value = convert_to_type(value, type) + return value + + def __get_type(self, key: Type, *, default=raise_on_missing): + value = self.__type_value.get(key) + if value is not None: + return value + key_str = self.__type_path.get(key) + if key_str is None: + if default is raise_on_missing: + raise ValueError(f"Key not found for type {key}") + return default + try: + value = self.get(key_str, key) + self.__type_value = value + except ValueError: + if default is raise_on_missing: + raise + value = default + return value + + def _merge(self, configs: Iterable[DictConfig]): + """Merge the current configuration with the given ones""" + self.c = cast(DictConfig, OmegaConf.merge(self.c, *configs)) + + def setup_configuration( + self, + conf: Union[DictConfig, dict, Any], + helpers: Dict[str, str] = {}, + *, + path: str = "", + ): + """Add a default configuration + + :param conf: The configuration to merge into the global configuration + :param helpers: Description of parameters used in argument parser helpers + :param path: The path to add the configuration to + """ + if isinstance(conf, type): + conf_type = conf + elif pydantic and issubclass(type(conf), pydantic.BaseModel): + conf_type = type(conf) + else: + conf_type = None + if conf_type: + # if already registered, set path to None + self.__type_path[conf_type] = None if conf_type in self.__type_path else path + if path and not path.endswith('.'): + path += "." + if isinstance(conf, str): + warnings.warn("provide a dict directly", DeprecationWarning) + created_config = OmegaConf.create(conf) + if not isinstance(created_config, DictConfig): + raise ValueError("The config is not a dict") + conf = created_config + if isinstance(conf, DictConfig): + config = self.__prepare_dictconfig(conf, path=path) + else: + created_config = self.__prepare_config(conf, path=path) + if not isinstance(created_config, DictConfig): + raise ValueError("Failed to convert to a DictConfig") + config = created_config + # add path and merge + if path: + config = self.__add_path(config, path.rstrip(".")) + self._merge([config]) + self.helpers.update(**helpers) + + 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), value) + for name, value in os.environ.items() + if name.startswith(prefixes) + ] + conf = OmegaConf.create({}) + for name, value in dotlist: + # TODO adapt name something.my_config from something.my.config + 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 + + def __prepare_dictconfig( + self, obj: DictConfig, path: str, recursive: bool = True + ) -> DictConfig: + sub_configs = [] + for k, v in obj.items_ex(resolve=False): + if not isinstance(k, str): + raise ValueError("Expecting only str instances in dict") + if recursive: + v = self.__prepare_config(v, path + k + ".") + if '.' in k: + obj.pop(k) + sub_configs.append(self.__add_path(v, k)) + if sub_configs: + obj = cast(DictConfig, OmegaConf.unsafe_merge(obj, *sub_configs)) + return obj + + def __prepare_config(self, obj, path): + if isinstance(obj, DictConfig): + return self.__prepare_dictconfig(obj, path) + if pydantic: + obj = self.__prepare_pydantic(obj, path) + if isinstance(obj, dict): + result = {} + changed = False + for k, v in obj.items(): + result[k] = nv = self.__prepare_config(v, path + k + ".") + changed |= v is not nv + if not changed: + result = obj + return self.__prepare_dictconfig(OmegaConf.create(result), path, recursive=False) + return obj + + def __prepare_pydantic(self, obj, path): + if isinstance(obj, pydantic.BaseModel): + # pydantic instance, prepare helpers + self.__prepare_pydantic(type(obj), path) + return obj.model_dump(mode="json") + # parse typing recursively for documentation + for t in typing.get_args(obj): + self.__prepare_pydantic(t, path) + # check if not a type + if not isinstance(obj, type): + return obj + # prepare documentation from types + if issubclass(obj, pydantic.BaseModel): + # pydantic type + defaults = {} + for k, field in obj.model_fields.items(): + check_type = True + if field.default is not pydantic.fields._Unset: + defaults[k] = field.default + check_type = not bool(defaults[k]) + elif field.is_required(): + defaults[k] = "???" + else: + defaults[k] = None + if desc := (field.description or field.title): + self.add_helper(path + k, desc) + if check_type and field.annotation: + self.__prepare_pydantic(field.annotation, path + k + ".") + return defaults + return None + + @staticmethod + def __add_path(config: Any, path: str) -> DictConfig: + for part in reversed(path.split(".")): + config = OmegaConf.create({part: config}) + return config diff --git a/alphaconf/internal/dotenv_vars.py b/alphaconf/internal/dotenv_vars.py new file mode 100644 index 0000000..3c662af --- /dev/null +++ b/alphaconf/internal/dotenv_vars.py @@ -0,0 +1,31 @@ +import logging +import os +from typing import Optional + +log = logging.getLogger(__name__) +_loaded: bool = False + + +def try_dotenv(load_dotenv: Optional[bool] = None): + """Load dotenv variables (optionally)""" + global _loaded + if load_dotenv is False or (_loaded and load_dotenv is None): + return + _loaded = True + try: + import dotenv + except ModuleNotFoundError: + if load_dotenv: + raise + log.debug('dotenv is not installed') + return + path = dotenv.find_dotenv(usecwd=True) + log.debug('Loading dotenv: %s', path or '(none)') + if not path: + return + dotenv.load_dotenv(path) + # check local overrides + path += '.local' + if os.path.isfile(path): + log.debug('Loading dotenv: %s', path) + dotenv.load_dotenv(path) diff --git a/alphaconf/internal/type_resolvers.py b/alphaconf/internal/type_resolvers.py index d5c18bc..dc7494d 100644 --- a/alphaconf/internal/type_resolvers.py +++ b/alphaconf/internal/type_resolvers.py @@ -3,6 +3,11 @@ from omegaconf import OmegaConf +try: + import pydantic +except ImportError: + pydantic = None # type: ignore + __doc__ = """Resolves types when reading values from the configuration. You can add values to TYPE_CONVERTER which is used in `alphaconf.get()`. @@ -33,6 +38,7 @@ def parse_bool(value) -> bool: 'read_strip': lambda s: read_text(s).strip(), 'read_bytes': lambda s: Path(s).expanduser().read_bytes(), } +_type = type # register resolved from strings for _name, _function in TYPE_CONVERTER.items(): @@ -47,5 +53,11 @@ def convert_to_type(value, type): :param type: A class or a callable used to convert the value :return: Result of the callable """ + if pydantic: + if issubclass(type, pydantic.BaseModel): + type.model_construct + return type.model_validate(value) + if isinstance(type, _type): + return pydantic.TypeAdapter(type).validate_python(value) type = TYPE_CONVERTER.get(type, type) return type(value) diff --git a/alphaconf/invoke.py b/alphaconf/invoke.py index bb6f0bb..ccf1e73 100644 --- a/alphaconf/invoke.py +++ b/alphaconf/invoke.py @@ -3,8 +3,8 @@ import invoke from omegaconf import OmegaConf -from . import run as _application_run -from .internal import Application, arg_parser +from .cli import run as _application_run +from .internal import application, arg_parser __doc__ = """Invoke wrapper for an application @@ -18,12 +18,14 @@ 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): +class InvokeApplication(application.Application): """Application that launched an invoke.Program""" def __init__(self, namespace: invoke.Collection, **properties) -> None: @@ -36,22 +38,22 @@ def __init__(self, namespace: invoke.Collection, **properties) -> None: ) def _handle_parsed_result(self): - if self.parsed.result: - return self.parsed.result.run(self) - return None + 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 + argv = [self.name, *self.parsed.rest] namespace = self.namespace - configuration = OmegaConf.to_object(self.configuration) + 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""" + """Create a new collection base on tasks in the variables""" return invoke.Collection(*[v for v in variables.values() if isinstance(v, invoke.Task)]) @@ -69,9 +71,12 @@ def run( _application_run(app.run_program, app=app) else: # Just configure the namespace and set the application - from . import get, set_application + import alphaconf + import alphaconf.logging_util - app.setup_configuration(arguments=False, load_dotenv=False, setup_logging=True) - set_application(app, merge=True) - ns.configure(get("")) + 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/alphaconf/logging_util.py b/alphaconf/logging_util.py index 49202ad..524fb33 100644 --- a/alphaconf/logging_util.py +++ b/alphaconf/logging_util.py @@ -3,7 +3,7 @@ import logging import traceback from logging import Formatter, LogRecord -from typing import Any, Callable +from typing import Any, Callable, Union try: import colorama @@ -37,6 +37,28 @@ _LOG_RECORD_FIELDS = set(logging.makeLogRecord({}).__dict__.keys()) +def setup_application_logging(configuration: Union[dict, None]) -> None: + """Setup logging + + Set the time to GMT, log key 'logging' from configuration or if none, base logging. + """ + import logging + + set_gmt() + log = logging.getLogger() + if configuration: + # Configure using the st configuration + import logging.config + + logging.config.dictConfig(configuration) + elif len(log.handlers) == 0: + # Default logging if not yet initialized + output = logging.StreamHandler() + output.setFormatter(ColorFormatter()) + log.addHandler(output) + log.setLevel(logging.INFO) + + def set_gmt(enable: bool = True): """Set GMT time for logging formatters diff --git a/demo.ipynb b/demo.ipynb index 58ace73..32b2da6 100644 --- a/demo.ipynb +++ b/demo.ipynb @@ -34,7 +34,6 @@ "\n", "positional arguments:\n", " key=value Configuration items\n", - " server Arguments for the demo\n", " show The name of the selection to show\n", " exception If set, raise an exception\n", "\u001b[0m" @@ -82,12 +81,10 @@ " logging:\n", " - default\n", " - none\n", + "exception: false\n", "server:\n", - " url: http://default\n", + " name: test_server\n", " user: ${oc.env:USER}\n", - " home: '~'\n", - "show: false\n", - "exception: false\n", "application:\n", " name: example\n", " version: '0.1'\n", @@ -113,11 +110,8 @@ "output_type": "stream", "text": [ "app: example\n", + "server.name test_server\n", "server.user: k\n", - "server.home /home/k\n", - "INFO:root:['init'] The app is running...\n", - "INFO:root:Just a log\n", - "INFO:root:['finished'] Application end.\n", "\u001b[0m" ] } @@ -136,11 +130,11 @@ "name": "stdout", "output_type": "stream", "text": [ - "INFO \u001b[32mApplication start (example-inv: InvokeApplication.run_program)\u001b[0m\n", + "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] and in alphaconf [4]\u001b[0m\n", - "INFO \u001b[32mApplication end.\u001b[0m\n", + "INFO \u001b[32mParam: [4]\u001b[0m\n", + "INFO \u001b[32mEnd.\u001b[0m\n", "\u001b[0m" ] } @@ -170,7 +164,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.8.10" + "version": "3.11.6" }, "orig_nbformat": 2 }, diff --git a/example-inv.py b/example-inv.py index 70bb1a2..fc9460e 100644 --- a/example-inv.py +++ b/example-inv.py @@ -10,14 +10,8 @@ def doit(ctx, param=None): """Some documentation...""" logging.info('Hello') # get the default configuration - logging.info('Backup: %s', alphaconf.configuration.get().backup) - # if we have a parameter, let's run some code within a context - # where we use this parameter in the configuration - if param: - with alphaconf.set(param=param): - logging.info('Param: [%s] and in alphaconf [%s]', param, alphaconf.get('param')) - else: - logging.warning('No parameter') + logging.info('Backup: %s', alphaconf.get("backup", default=None)) + logging.info('Param: [%s]', param) # add some default configuration and run/configure invoke's namespace diff --git a/example-simple.py b/example-simple.py index 7fad697..7af35de 100755 --- a/example-simple.py +++ b/example-simple.py @@ -1,26 +1,28 @@ #!/usr/bin/env python3 import logging -from pathlib import Path +from typing import Optional -import alphaconf +from pydantic import BaseModel, Field + +import alphaconf.cli import alphaconf.logging_util + +class Opts(BaseModel): + show: Optional[str] = Field(None, description="The name of the selection to show") + exception: bool = Field(False, description="If set, raise an exception") + + # adding a default configuration # these will be merged with the application +alphaconf.setup_configuration(Opts) alphaconf.setup_configuration( - """ -server: - url: http://default - user: ${oc.env:USER} - home: "~" -show: false -exception: false -""", { - "server": "Arguments for the demo", - "show": "The name of the selection to show", - "exception": "If set, raise an exception", - }, + "server": { + "name": "test_server", + "user": "${oc.env:USER}", + } + } ) @@ -28,10 +30,10 @@ def main(): """Simple demo of alphaconf""" # get the application name from the configuration - print('app:', alphaconf.configuration.get().application.name) + print('app:', alphaconf.get("application.name")) # shortcut version to get a configuration value + print('server.name', alphaconf.get('server.name')) print('server.user:', alphaconf.get('server.user')) - print('server.home', alphaconf.get('server.home', Path)) # you can set additional dynamic values in the logging context_value = ['init'] @@ -43,11 +45,11 @@ def main(): logging.info('Just a log') # show configuration - value = alphaconf.get('show', str) - if value and (value := alphaconf.get(value)): + value = alphaconf.get('show', str, default=None) + if value and (value := alphaconf.get(value, default=None)): print(value) # log an exception if we have it in the configuration - if alphaconf.get('exception'): + if alphaconf.get('exception', default=False): try: raise RuntimeError("Asked to raise something") except Exception: @@ -57,7 +59,7 @@ def main(): if __name__ == '__main__': # running with explicit parameters - alphaconf.run( + alphaconf.cli.run( main, name='example', version='0.1', diff --git a/example-typed.py b/example-typed.py index 4ac68da..ece0717 100755 --- a/example-typed.py +++ b/example-typed.py @@ -1,25 +1,40 @@ #!/usr/bin/env python3 import logging -from dataclasses import dataclass +from datetime import date +from pathlib import Path from typing import Optional -from omegaconf import OmegaConf +from pydantic import BaseModel, Field -import alphaconf +import alphaconf.cli -@dataclass -class MyConfiguration: - name: Optional[str] = None +class Conn(BaseModel): + url: str = Field("http://github.com", title="Some URL") + home: Path = Path("~") -alphaconf.setup_configuration({'c': OmegaConf.structured(MyConfiguration)}) +class MyConfiguration(BaseModel): + name: Optional[str] = Field(None, title="Some name to show") + some_date: Optional[date] = None + connection: Optional[Conn] = None + + +alphaconf.setup_configuration(MyConfiguration, path="c") def main(): """Typed configuration example""" logging.info('Got configuration name: %s', alphaconf.get('c.name')) + c = alphaconf.get(MyConfiguration) + logging.info("Found configuration object: %s", c) + if c.connection: + logging.info( + 'connection.home: %s (%s)', + alphaconf.get('c.connection.home', default='unset'), + c.connection.home, + ) if __name__ == '__main__': - alphaconf.run(main) + alphaconf.cli.run(main) diff --git a/pre-commit b/pre-commit index 5a418a1..a8903c1 100755 --- a/pre-commit +++ b/pre-commit @@ -12,12 +12,20 @@ pre_commit() { mypy . } +format() { + black . + isort . +} + # Commands case "${1:-run}" in run|lint) pre_commit echo "All good to commit" ;; + format) + format + ;; install) echo "Installing pre-commit" cd .git/hooks @@ -29,5 +37,5 @@ case "${1:-run}" in ;; *) echo "Invalid argument: $*" - echo "Supported options: lint, install, uninstall" + echo "Supported options: lint, format, install, uninstall" esac diff --git a/pyproject.toml b/pyproject.toml index 0eda521..9580fbc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,9 +11,9 @@ name = "alphaconf" dynamic = ["version"] description = "Write simple scripts leveraging omegaconf" readme = "README.md" -keywords = ["configuration", "omegaconf", "script"] +keywords = ["configuration", "omegaconf", "pydantic", "script"] license = {text = "BSD License"} -requires-python = ">=3.8" +requires-python = ">=3.9" classifiers = [ # https://pypi.org/pypi?%3Aaction=list_classifiers "Programming Language :: Python :: 3", @@ -23,6 +23,7 @@ classifiers = [ ] dependencies = [ "omegaconf>=2", + "pydantic>=2", ] [project.optional-dependencies] diff --git a/requirements.txt b/requirements.txt index 1aca565..a0fcc0b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,6 +11,7 @@ types-toml # Libraries omegaconf>=2 colorama +pydantic>=2 python-dotenv invoke toml diff --git a/tests/conftest.py b/tests/conftest.py index 49f015d..6b18a67 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,10 +1,17 @@ import pytest import alphaconf +from alphaconf.internal.configuration import Configuration @pytest.fixture(autouse=True) -def reset_context_configuration(): - config = alphaconf.configuration.get() +def reset_configuration(): + alphaconf._application = None + old = alphaconf._global_configuration + alphaconf._global_configuration = config = Configuration(parent=old) + alphaconf.setup_configuration = config.setup_configuration + alphaconf.get = config.get yield - alphaconf.configuration.set(config) + alphaconf._global_configuration = old + alphaconf.setup_configuration = old.setup_configuration + alphaconf.get = old.get diff --git a/tests/test_alphaconf.py b/tests/test_alphaconf.py index 694a0ff..3544057 100644 --- a/tests/test_alphaconf.py +++ b/tests/test_alphaconf.py @@ -1,7 +1,9 @@ +import os + import pytest from omegaconf import DictConfig, OmegaConf -import alphaconf +import alphaconf.cli @pytest.fixture(scope='function') @@ -10,7 +12,7 @@ def application(): def test_default_configuration(): - config = alphaconf.configuration.get() + config = alphaconf._global_configuration.c assert isinstance(config, DictConfig) assert 'base' in config assert 'logging' in config @@ -18,21 +20,21 @@ def test_default_configuration(): def test_run(): result = 'result' - r = alphaconf.run(lambda: result, arguments=False, should_exit=False) + r = alphaconf.cli.run(lambda: result, arguments=False, should_exit=False) assert r is result def test_run_application_init(): name = 'testinit' - assert alphaconf.get('application') is None - r = alphaconf.run(lambda: alphaconf.get('application.name'), arguments=False, name=name) + assert alphaconf.get('application', default=None) is None + r = alphaconf.cli.run(lambda: alphaconf.get('application.name'), arguments=False, name=name) assert name == r def test_run_application_help(capsys): alphaconf.setup_configuration({'helptest': 1}, {'helptest': 'HELPER_TEST'}) application = alphaconf.Application(name='myapp', description='my test description') - r = alphaconf.run(lambda: 1, app=application, arguments=['--help'], should_exit=False) + r = alphaconf.cli.run(lambda: 1, app=application, arguments=['--help'], should_exit=False) assert r is None captured = capsys.readouterr() out = captured.out.splitlines() @@ -40,19 +42,21 @@ def test_run_application_help(capsys): assert 'usage:' in out[0] assert application.name in out[0] desc = '\n'.join(out[1:5]) - assert application.description in desc + assert application.properties['description'] in desc assert 'HELPER_TEST' in captured.out def test_run_application_version(capsys, application): - alphaconf.run(lambda: 'n', app=application, arguments=['--version'], should_exit=False) + alphaconf.cli.run(lambda: 'n', app=application, arguments=['--version'], should_exit=False) captured = capsys.readouterr() out = captured.out - assert (application.name + ' ' + application.version) in out + assert (application.name + ' ' + application.properties['version']) in out def test_run_application_show_configuration(capsys, application): - alphaconf.run(lambda: 'n', app=application, arguments=['--configuration'], should_exit=False) + alphaconf.cli.run( + lambda: 'n', app=application, arguments=['--configuration'], should_exit=False + ) captured = capsys.readouterr() out = captured.out data = OmegaConf.to_container(OmegaConf.create(out)) @@ -62,22 +66,20 @@ def test_run_application_show_configuration(capsys, application): def test_run_application_set_argument(): - r = alphaconf.run(lambda: alphaconf.get('a.b'), arguments=['a.b=36']) + r = alphaconf.cli.run(lambda: alphaconf.get('a.b'), arguments=['a.b=36']) assert r == 36 def test_run_application_select_logging(): - log = alphaconf.run(lambda: alphaconf.get('logging'), arguments=['--select', 'logging=none']) + log = alphaconf.cli.run( + lambda: alphaconf.get('logging'), arguments=['--select', 'logging=none'] + ) assert isinstance(log, dict) def test_set_application(application): - token = alphaconf.configuration.set(OmegaConf.create()) - try: - alphaconf.set_application(application) - assert alphaconf.configuration.get() == application.configuration - finally: - alphaconf.configuration.reset(token) + alphaconf.set_application(application) + assert alphaconf._application is application def test_setup_configuration(): @@ -91,19 +93,6 @@ def test_setup_configuration_invalid(): with pytest.raises(ValueError): # invalid configuration (must be non-empty) alphaconf.setup_configuration(None) - with pytest.raises(ValueError): - # invalid helper - alphaconf.setup_configuration({'invalid': 5}, helpers={'help': 'help1'}) - - -def test_set(): - # test that the set() is active only within the block - value = '124' - default = '125-def' - assert alphaconf.get('value', default=default) is default - with alphaconf.set(value=value): - assert alphaconf.get('value') is value - assert alphaconf.get('value', default=default) is default def test_secret_masks(): @@ -123,3 +112,28 @@ def test_secret_masks(): finally: alphaconf.SECRET_MASKS.clear() alphaconf.SECRET_MASKS.extend(masks) + + +def test_app_setup_configuration(application): + application.setup_configuration( + arguments=['a=x'], load_dotenv=False, env_prefixes=False, resolve_configuration=False + ) + assert application.configuration.get('a') == 'x' + + +def test_app_environ(application): + alphaconf.setup_configuration({"testmyenv": {"x": 1}}) + os.environ.update( + { + 'XXX': 'not set', + 'TESTMYENV_X': 'overwrite', + 'TESTMYENV_Y': 'new', + } + ) + application.setup_configuration(load_dotenv=False, env_prefixes=True) + config = application.configuration + with pytest.raises(ValueError): + # XXX should not be loaded + config.get('xxx') + assert config.get('testmyenv.x') == 'overwrite' + assert config.get('testmyenv.y') == 'new' diff --git a/tests/test_arg_parser.py b/tests/test_arg_parser.py index 36bbd12..e148ab3 100644 --- a/tests/test_arg_parser.py +++ b/tests/test_arg_parser.py @@ -37,7 +37,7 @@ def test_application_version(parser): def test_parse_arguments(parser): other_arguments = ['other', 'arguments'] - r = parser.parse_args(['hello=world', 'test=123', '--'] + other_arguments) + r = parser.parse_args(['hello=world', 'test=123', '--', *other_arguments]) print(r) assert r.result is None conf = OmegaConf.merge(*r.configurations()) diff --git a/tests/test_configuration.py b/tests/test_configuration.py new file mode 100644 index 0000000..8fa81ce --- /dev/null +++ b/tests/test_configuration.py @@ -0,0 +1,103 @@ +from pathlib import Path + +import pytest +from omegaconf import DictConfig + +from alphaconf import Configuration + + +@pytest.fixture(scope='function') +def config(): + c = { + 'a': {'b': 3}, + 'root': 'R', + 'b': True, + 'num': 5, + 'home': '/home', + } + conf = Configuration() + conf.setup_configuration(c) + return conf + + +@pytest.fixture(scope='function') +def config_req(config): + config.setup_configuration({'req': '???'}) + return config + + +@pytest.mark.parametrize( + "key,expected", + [ + ('a', {'b': 3}), + ('a.b', 3), + ('b', True), + ('num', 5), + ], +) +def test_select_dict(config, key, expected): + assert config.get(key) == expected + + +def test_default(config): + assert config.get('nonexistingkey', default=3) == 3 + + +def test_cast(config): + # cast bool into int + assert config.get('b') is True + assert config.get('b', int) == 1 + # cast Path + assert isinstance(config.get('home', Path), Path) + + +def test_cast_inexisting(config): + assert config.get('nonexistingkey', int, default=None) is None + assert config.get('nonexistingkey', int, default="abc") == "abc" + + +def test_cast_omega(config): + conf = config.get('', DictConfig) + assert isinstance(conf, DictConfig) + assert conf.a == {'b': 3} + assert config.get('', DictConfig) is conf + + +def test_select_empty(config): + result = config.get("") + assert all(k in result for k in config.c) + + +def test_select_required(config): + assert config.get('z', default=None) is None + with pytest.raises(ValueError): + print(config.get('z')) + assert config.get('z', default='a') == 'a' + + +def test_select_required_incomplete(config_req): + # when we have a default, return it + assert config_req.get('req', default='def') == 'def' + # when required, raise missing + with pytest.raises(ValueError): + print(config_req.get('req')) + + +def test_config_setup_dots(config): + config.setup_configuration( + { + 'a.b': { + 'c.d': 1, + 'two': 2, + 'c.x': 'x', + }, + } + ) + assert config.get('a.b.c.d') == 1 + assert config.get('a.b.c.x') == 'x' + assert config.get('a.b.two') == 2 + + +def test_config_setup_path(config): + config.setup_configuration({'test': 954}, path='a.b') + assert config.get('a.b.test') == 954 diff --git a/tests/test_configuration_typed.py b/tests/test_configuration_typed.py new file mode 100644 index 0000000..3f1808d --- /dev/null +++ b/tests/test_configuration_typed.py @@ -0,0 +1,100 @@ +from datetime import date, datetime +from pathlib import Path +from typing import Optional + +import pydantic +import pytest + +from alphaconf import Configuration + + +class Person(pydantic.BaseModel): + first_name: str + last_name: str + + @property + def full_name(self): + return ' '.join([self.first_name, self.last_name]) + + +class TypedConfig(pydantic.BaseModel): + x_path: Path = Path('~') + x_date: Optional[date] = None + x_datetime: Optional[datetime] = None + x_name: str = pydantic.Field('me', description="Some name") + x_num: int = 0 + x_person: Optional[Person] = None + + +@pytest.fixture(scope='function') +def config_typed(): + c = Configuration() + c.setup_configuration(TypedConfig) + return c + + +@pytest.fixture(scope='function') +def config_changed(config_typed): + config_typed.setup_configuration( + { + 'x_num': 1, + 'x_name': 'test', + 'x_date': '2023-05-06', + 'x_datetime': '2023-08-06 00:00:00', + } + ) + return config_typed + + +@pytest.mark.parametrize( + "key,expected", + [ + ('x_num', 0), + ('x_name', "me"), + ('x_date', None), + ('x_person', None), + ], +) +def test_get_default(config_typed, key, expected): + v = config_typed.get(TypedConfig) + assert getattr(v, key) == expected + + +@pytest.mark.parametrize( + "key,expected", + [ + ('x_num', 1), + ('x_name', "test"), + ('x_date', date(2023, 5, 6)), + ('x_datetime', datetime(2023, 8, 6)), + ('x_person', None), + ], +) +def test_get_changed(config_changed, key, expected): + v = config_changed.get(TypedConfig) + assert getattr(v, key) == expected + + +def test_get_path(config_typed): + v = config_typed.get(TypedConfig) + assert isinstance(v.x_path, Path) + + +def test_set_person(config_typed): + config_typed.setup_configuration( + { + 'x_person': { + 'first_name': 'A', + 'last_name': 'B', + } + } + ) + person = config_typed.get('x_person', Person) + assert person + assert person.full_name == 'A B' + + +def test_set_person_type(config_typed): + config_typed.setup_configuration(Person(first_name='A', last_name='T'), path='x_person') + person = config_typed.get(Person) + assert person.full_name == 'A T' diff --git a/tests/test_logging.py b/tests/test_logging.py index a790030..ddddf8b 100644 --- a/tests/test_logging.py +++ b/tests/test_logging.py @@ -8,13 +8,15 @@ @pytest.fixture(scope='function') def application(): - return alphaconf.Application(name='test_logging') + app = alphaconf.Application(name='test_logging') + app.setup_configuration(load_dotenv=False) + return app @pytest.fixture(scope='function') def log(application): - application.setup_configuration(arguments=False, setup_logging=True) alphaconf.set_application(application) + alphaconf.logging_util.setup_application_logging(application.configuration.get("logging")) return logging.getLogger() @@ -24,8 +26,8 @@ def test_log_ok(log, caplog): log.warning('twarn') print(caplog.records) assert len(caplog.records) == 2 # should not capture debug by default - assert 'tinfo' == caplog.records[0].message - assert 'twarn' == caplog.records[1].message + assert caplog.records[0].message == 'tinfo' + assert caplog.records[1].message == 'twarn' def test_log_exception(log, caplog): diff --git a/tests/test_select_and_type.py b/tests/test_select_and_type.py deleted file mode 100644 index b65784b..0000000 --- a/tests/test_select_and_type.py +++ /dev/null @@ -1,90 +0,0 @@ -import pathlib - -import pytest - -from alphaconf import DictConfig, MissingMandatoryValue, select - - -@pytest.fixture(scope='function') -def container(): - return { - 'a': {'b': 3}, - 'root': 'R', - 'b': True, - 'home': '/home', - } - - -@pytest.fixture(scope='function') -def container_incomplete(container): - return DictConfig( - { - 'a': container['a'], - 'req': '???', - } - ) - - -@pytest.mark.parametrize( - "key,expected", - [ - ('a', {'b': 3}), - ('a.b', 3), - ('z', None), - ('b', True), - ], -) -def test_select_dict(container, key, expected): - assert select(container, key) == expected - - -def test_default(container): - assert select(container, 'nonexistingkey', default=3) == 3 - - -def test_cast(container): - # cast bool into int - assert select(container, 'b', int) == 1 - # cast Path - assert isinstance(select(container, 'home', pathlib.Path), pathlib.Path) - # cast inexisting and default - assert select(container, 'nonexistingkey', int) is None - assert select(container, 'nonexistingkey', int, default="abc") == "abc" - - -def test_cast_omega(container): - conf = select(container, '', DictConfig) - assert isinstance(conf, DictConfig) - assert conf.a == {'b': 3} - assert select(conf, '', DictConfig) is conf - - -def test_select_empty(container): - assert select(container, '') == container - - -def test_select_required(container): - cont = DictConfig(container) - assert select(cont, 'z') is None - with pytest.raises(MissingMandatoryValue): - print(select(cont, 'z', required=True)) - assert select(cont, 'z', required=True, default='a') == 'a' - - -def test_select_required_incomplete(container_incomplete): - cont = container_incomplete - # when we have a default, return it - assert select(cont, 'req', default='def') == 'def' - # when required, raise missing - with pytest.raises(MissingMandatoryValue): - print(select(cont, 'req', required=True)) - # when required, even if a default is given, if req=???, raise - with pytest.raises(MissingMandatoryValue): - print(select(cont, 'req', required=True, default='def')) - - -def test_select_from_any(): - # for now, we allow selecting from anything that can be a container - # OmegaConf.create() return the value to be selected - assert select([1, 2, 3], '[0]') == 1 - assert select("[1, 2, 3]", '[0]') == 1 From af440bde53a89d2107f7891909ac645059e50d67 Mon Sep 17 00:00:00 2001 From: Krzysztof Magusiak Date: Mon, 27 Nov 2023 21:43:55 +0100 Subject: [PATCH 3/4] Typing and secrets (#22) --- README.md | 58 +++++++++++++++++----------- alphaconf/__init__.py | 8 ++-- alphaconf/interactive.py | 4 +- alphaconf/internal/application.py | 8 ++-- alphaconf/internal/configuration.py | 37 ++++++++++++++++-- alphaconf/internal/type_resolvers.py | 20 ++++++---- pyproject.toml | 2 +- tests/test_configuration.py | 33 ++++++++++++++++ 8 files changed, 125 insertions(+), 45 deletions(-) diff --git a/README.md b/README.md index 2cb9fc1..4e1a194 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ alphaconf.setup_configuration({ def main(): log = logging.getLogger() log.info('server.url:', alphaconf.get('server.url')) - log.info('has server.user:', alphaconf.get('server.user', bool)) + log.info('has server.user:', alphaconf.get('server.user', bool, default=False)) if __name__ == '__main__': alphaconf.cli.run(main) @@ -76,17 +76,17 @@ Finally, the configuration is fully resolved and logging is configured. ## Configuration templates and resolvers -[OmegaConf]'s resolvers may be used as configuration values. -For example, `${oc.env:USER,me}` would resolve to the environment variable -USER with a default value "me". -Similarly, `${oc.select:path}` will resolve to another configuration value. - -Additional resolvers are added to read file contents. -These are the same as type casts: read_text, read_strip, read_bytes. --- TODO use secrets for v1 - -The select is used to build multiple templates for configurations by providing -base configurations. +Configuration values are resolved by [OmegaConf]. +Some of the resolvers (standard and custom): +- `${oc.env:USER,me}`: resolve the environment variable USER + with a default value "me" +- `${oc.select:config_path}`: resolve to another configuration value +- `${read_text:file_path}`: read text contents of a file as `str` +- `${read_bytes:file_path}`: read contents of a file as `bytes` +- `${read_strip:file_path}`: read text contents of a file as strip spaces + +The *oc.select* is used to build multiple templates for configurations +by providing base configurations. An argument `--select key=template` is a shortcut for `key=${oc.select:base.key.template}`. So, `logging: ${oc.select:base.logging.default}` resolves to the configuration @@ -96,20 +96,31 @@ dict defined in base.logging.default and you can select it using ## Configuration values and integrations ### Typed-configuration --- TODO update to pydantic -You can use *omegaconf* with *dataclasses* to specify which values are -enforced in the configuration. -Alternatively, the *get* method can receive a data type or a function -which will parse the value. -By default, bool, str, Path, DateTime, etc. are supported. +You can use [OmegaConf] with [pydantic] to *get* typed values. +```python +class MyConf(pydantic.BaseModel): + value: int = 0 + + def build(self): + # use as a factory pattern to create more complex objects + return self.value * 2 + +# setup the configuration +alphaconf.setup_configuration(MyConf, path='a') +# read the value +alphaconf.get('a', MyConf) +v = alphaconf.get(MyConf) # because it's registered as a type +``` ### Secrets When showing the configuration, by default configuration keys which are secrets, keys or passwords will be masked. -Another good practice is to have a file containing the password which -you can retrieve using `alphaconf.get('secret_file', 'read_strip')`. +You can read values or passwords from files, by using the template +`${read_strip:/path_to_file}` +or, more securely, read the file in the code +`alphaconf.get('secret_file', Path).read_text().strip()`. ### Invoke integration @@ -125,9 +136,10 @@ alphaconf.invoke.run(__name__, ns) ``` ## Way to 1.0 -- Secret management -- Install completions for bash -- Run a function after importing the module +- Run function `@alphaconf.inject` +- Run a specific function `alphaconf.cli.run_module()`: + find functions and parse their 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 fcffc63..c05cc25 100644 --- a/alphaconf/__init__.py +++ b/alphaconf/__init__.py @@ -1,6 +1,6 @@ import re import warnings -from typing import Callable, Optional, Sequence, TypeVar, Union +from typing import Callable, MutableSequence, Optional, Sequence, TypeVar, Union from .frozendict import frozendict # noqa: F401 (expose) from .internal.application import Application @@ -26,9 +26,11 @@ """ -SECRET_MASKS = [ +SECRET_MASKS: MutableSequence[Callable] = [ # mask if contains a kind of secret and it's not in a file - re.compile(r'.*(key|password|secret)s?(?!_file)(_|$)|^private(_key|$)').match, + re.compile( + r'.*(key|password|secret)s?(?!_file)(?!_path)(_|$)|^(authentication|private)(_key|$)' + ).match, ] """A list of functions which given a key indicate whether it's a secret""" diff --git a/alphaconf/interactive.py b/alphaconf/interactive.py index 004b654..53216a7 100644 --- a/alphaconf/interactive.py +++ b/alphaconf/interactive.py @@ -16,10 +16,10 @@ def mount(configuration_paths: List[str] = [], setup_logging: bool = True): application.setup_configuration(configuration_paths=configuration_paths) set_application(application) if setup_logging: - import logging_util + from . import logging_util logging_util.setup_application_logging( - application.configuration.get('logging'), default=None + application.configuration.get('logging', default=None) ) logging.info('Mounted interactive application') diff --git a/alphaconf/internal/application.py b/alphaconf/internal/application.py index 2815921..b4e3691 100644 --- a/alphaconf/internal/application.py +++ b/alphaconf/internal/application.py @@ -3,7 +3,7 @@ import os import sys import uuid -from typing import Iterable, List, MutableMapping, Optional, Tuple, Union, cast +from typing import Callable, Iterable, List, MutableMapping, Optional, Tuple, Union, cast from omegaconf import DictConfig, OmegaConf @@ -24,7 +24,7 @@ class Application: def __init__( self, *, - name=None, + name: Optional[str] = None, **properties, ) -> None: """Initialize the application. @@ -237,7 +237,7 @@ def masked_configuration( return config @staticmethod - def __mask_config(obj, check, replace, path=''): + def __mask_config(obj, check: Callable[[str], bool], replace: Callable, path: str = ''): """Alter the configuration dict :param config: The value to mask @@ -264,7 +264,7 @@ def __mask_config(obj, check, replace, path=''): ] return obj - def print_help(self, *, arguments=True): + def print_help(self, *, arguments: bool = True): """Print the help message Set the arguments to False to disable printing them.""" prop = self.properties diff --git a/alphaconf/internal/configuration.py b/alphaconf/internal/configuration.py index 4f958f1..8da1d1e 100644 --- a/alphaconf/internal/configuration.py +++ b/alphaconf/internal/configuration.py @@ -7,6 +7,7 @@ Any, Dict, Iterable, + List, MutableMapping, Optional, Type, @@ -62,7 +63,7 @@ def get( def get( self, key: str, - type: Union[str, Type[T], None] = None, + type: Union[str, None] = None, *, default: Any = raise_on_missing, ) -> Any: @@ -146,6 +147,7 @@ def setup_configuration( if conf_type: # if already registered, set path to None self.__type_path[conf_type] = None if conf_type in self.__type_path else path + self.__type_value.pop(conf_type, None) if path and not path.endswith('.'): path += "." if isinstance(conf, str): @@ -164,7 +166,9 @@ def setup_configuration( # add path and merge if path: config = self.__add_path(config, path.rstrip(".")) + helpers = {path + k: v for k, v in helpers.items()} self._merge([config]) + # helpers self.helpers.update(**helpers) def add_helper(self, key, description): @@ -178,13 +182,13 @@ def from_environ(self, prefixes: Iterable[str]) -> DictConfig: trans = str.maketrans('_', '.', '"\\=') prefixes = tuple(prefixes) dotlist = [ - (name.lower().translate(trans), value) + (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: - # TODO adapt name something.my_config from something.my.config + name = Configuration._find_name(name.split('.'), self.c) try: conf.merge_with_dotlist([f"{name}={value}"]) except YAMLError: @@ -192,6 +196,25 @@ def from_environ(self, prefixes: Iterable[str]) -> DictConfig: OmegaConf.update(conf, name, value) return conf + @staticmethod + def _find_name(parts: List[str], conf: DictConfig) -> str: + """Find a name from parts, by trying joining with '.' (default) or '_'""" + if len(parts) < 2: + return "".join(parts) + name = "" + for next_offset, part in enumerate(parts, 1): + if name: + name += "_" + name += part + if name in conf.keys(): + sub_conf = conf.get(name) + if next_offset == len(parts): + return name + elif isinstance(sub_conf, DictConfig): + return name + "." + Configuration._find_name(parts[next_offset:], sub_conf) + return ".".join([name, *parts[next_offset:]]) + return ".".join(parts) + def __prepare_dictconfig( self, obj: DictConfig, path: str, recursive: bool = True ) -> DictConfig: @@ -248,9 +271,15 @@ def __prepare_pydantic(self, obj, path): defaults[k] = "???" else: defaults[k] = None + # description if desc := (field.description or field.title): self.add_helper(path + k, desc) - if check_type and field.annotation: + # check the type + if field.annotation == pydantic.SecretStr: + from alphaconf import SECRET_MASKS + + SECRET_MASKS.append(lambda s: s == path) + elif check_type and field.annotation: self.__prepare_pydantic(field.annotation, path + k + ".") return defaults return None diff --git a/alphaconf/internal/type_resolvers.py b/alphaconf/internal/type_resolvers.py index dc7494d..be49a47 100644 --- a/alphaconf/internal/type_resolvers.py +++ b/alphaconf/internal/type_resolvers.py @@ -34,11 +34,11 @@ def parse_bool(value) -> bool: datetime.date: lambda s: datetime.datetime.strptime(s, '%Y-%m-%d').date(), datetime.time: datetime.time.fromisoformat, Path: lambda s: Path(s).expanduser(), + str: lambda v: str(v), 'read_text': read_text, 'read_strip': lambda s: read_text(s).strip(), 'read_bytes': lambda s: Path(s).expanduser().read_bytes(), } -_type = type # register resolved from strings for _name, _function in TYPE_CONVERTER.items(): @@ -50,14 +50,18 @@ def convert_to_type(value, type): """Converts a value to the given type. :param value: Any value - :param type: A class or a callable used to convert the value + :param type: A class used to convert the value :return: Result of the callable """ + if isinstance(type, str): + return TYPE_CONVERTER[type](value) + # assert isinstance(type, globals().type) + if pydantic and issubclass(type, pydantic.BaseModel): + return type.model_validate(value) + if isinstance(value, type): + return value + if type in TYPE_CONVERTER: + return TYPE_CONVERTER[type](value) if pydantic: - if issubclass(type, pydantic.BaseModel): - type.model_construct - return type.model_validate(value) - if isinstance(type, _type): - return pydantic.TypeAdapter(type).validate_python(value) - type = TYPE_CONVERTER.get(type, type) + return pydantic.TypeAdapter(type).validate_python(value) return type(value) diff --git a/pyproject.toml b/pyproject.toml index 9580fbc..af22d06 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ requires-python = ">=3.9" classifiers = [ # https://pypi.org/pypi?%3Aaction=list_classifiers "Programming Language :: Python :: 3", - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "License :: OSI Approved :: BSD License", "Environment :: Console", ] diff --git a/tests/test_configuration.py b/tests/test_configuration.py index 8fa81ce..b9386f0 100644 --- a/tests/test_configuration.py +++ b/tests/test_configuration.py @@ -14,6 +14,7 @@ def config(): 'b': True, 'num': 5, 'home': '/home', + 'with_underscore': "/_\\", } conf = Configuration() conf.setup_configuration(c) @@ -83,6 +84,38 @@ def test_select_required_incomplete(config_req): print(config_req.get('req')) +@pytest.mark.parametrize( + "name,expected", + [ + ('a.b', 'a.b'), + ('unknown', 'unknown'), + ('a.b.zz', 'a.b.zz'), + ('b', 'b'), + ], +) +def test_env_find_name_simple(config, name, expected): + assert Configuration._find_name(name.split('.'), config.c) == expected + + +def test_env_find_name_complex(): + config = Configuration() + config.setup_configuration( + { + 'a': {'b': 1}, + 'my_test': {'a': 2}, + 'test_test': { + 'x': 3, + 'my_test': 4, + }, + } + ) + c = config.c + assert Configuration._find_name(['a', 'b'], c) == 'a.b' + assert Configuration._find_name(['my', 'test'], c) == 'my_test' + assert Configuration._find_name(['my', 'test', 'a'], c) == 'my_test.a' + assert Configuration._find_name(['test', 'test', 'my', 'test'], c) == 'test_test.my_test' + + def test_config_setup_dots(config): config.setup_configuration( { From a93f0e140f91ed4496042cae0663df450fffe72a Mon Sep 17 00:00:00 2001 From: Krzysztof Magusiak Date: Mon, 4 Dec 2023 22:33:17 +0100 Subject: [PATCH 4/4] @inject decorator (#23) --- README.md | 36 ++++++++--- alphaconf/inject.py | 92 ++++++++++++++++++++++++++++ alphaconf/internal/configuration.py | 49 +++++++-------- alphaconf/internal/type_resolvers.py | 24 +++++--- example-typed.py | 2 +- pyproject.toml | 2 +- tests/test_alphaconf.py | 6 +- tests/test_configuration.py | 7 ++- tests/test_configuration_typed.py | 2 +- tests/test_inject.py | 81 ++++++++++++++++++++++++ 10 files changed, 250 insertions(+), 51 deletions(-) create mode 100644 alphaconf/inject.py create mode 100644 tests/test_inject.py diff --git a/README.md b/README.md index 4e1a194..8fef71e 100644 --- a/README.md +++ b/README.md @@ -66,10 +66,9 @@ Then configuration is built from: - `PYTHON_ALPHACONF` environment variable may contain a path to load - configuration files from configuration directories (using application name) - environment variables based on key prefixes, - except "BASE" and "PYTHON"; + except "BASE" and "PYTHON"; \ if you have a configuration key "abc", all environment variables starting - with "ABC_" will be loaded where keys are converted to lower case and "_" - to ".": "ABC_HELLO=a" would set "abc.hello=a" + with "ABC_" will be loaded, for example "ABC_HELLO=a" would set "abc.hello=a" - key-values from the program arguments Finally, the configuration is fully resolved and logging is configured. @@ -104,10 +103,11 @@ class MyConf(pydantic.BaseModel): def build(self): # use as a factory pattern to create more complex objects + # for example, a connection to the database return self.value * 2 # setup the configuration -alphaconf.setup_configuration(MyConf, path='a') +alphaconf.setup_configuration(MyConf, prefix='a') # read the value alphaconf.get('a', MyConf) v = alphaconf.get(MyConf) # because it's registered as a type @@ -122,6 +122,29 @@ You can read values or passwords from files, by using the template or, more securely, read the file in the code `alphaconf.get('secret_file', Path).read_text().strip()`. +### Inject parameters + +We can inject default values to functions from the configuration. +Either one by one, where we can map a factory function or a configuration key. +Or inject all automatically base on the parameter name. + +```python +from alphaconf.inject import inject, inject_auto + +@inject('name', 'application.name') +@inject_auto(ignore={'name'}) +def main(name: str, example=None): + pass + +# similar to +def main(name: str=None, example=None): + if name is None: + name = alphaconf.get('application.name', str) + if example is None: + example = alphaconf.get('example', default=example) + ... +``` + ### Invoke integration Just add the lines below to parameterize invoke. @@ -136,9 +159,8 @@ alphaconf.invoke.run(__name__, ns) ``` ## Way to 1.0 -- Run function `@alphaconf.inject` -- Run a specific function `alphaconf.cli.run_module()`: - find functions and parse their args +- 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/ diff --git a/alphaconf/inject.py b/alphaconf/inject.py new file mode 100644 index 0000000..ffd8620 --- /dev/null +++ b/alphaconf/inject.py @@ -0,0 +1,92 @@ +import functools +import inspect +from typing import Any, Callable, Dict, Optional, Union + +import alphaconf + +from .internal.type_resolvers import type_from_annotation + +__all__ = ["inject", "inject_auto"] + + +class ParamDefaultsFunction: + """Function wrapper that injects default parameters""" + + _arg_factory: Dict[str, Callable[[], Any]] + + def __init__(self, func: Callable): + self.func = func + self.signature = inspect.signature(func) + self._arg_factory = {} + + def bind(self, name: str, factory: Callable[[], Any]): + self._arg_factory[name] = factory + + def __call__(self, *a, **kw): + args = self.signature.bind_partial(*a, **kw).arguments + kw.update( + {name: factory() for name, factory in self._arg_factory.items() if name not in args} + ) + return self.func(*a, **kw) + + @staticmethod + def wrap(func) -> "ParamDefaultsFunction": + if isinstance(func, ParamDefaultsFunction): + return func + return functools.wraps(func)(ParamDefaultsFunction(func)) + + +def getter( + key: str, ktype: Optional[type] = None, *, param: Optional[inspect.Parameter] = None +) -> Callable[[], Any]: + """Factory function that calls alphaconf.get + + The parameter from the signature can be given to extract the type to cast to + and whether the configuration value is optional. + + :param key: The key using in alphaconf.get + :param ktype: Type to cast to + :param param: The parameter object from the signature + """ + if ktype is None and param and (ptype := param.annotation) is not param.empty: + ktype = next(type_from_annotation(ptype), None) + if param is not None and param.default is not param.empty: + xparam = param + return ( + lambda: xparam.default + if (value := alphaconf.get(key, ktype, default=None)) is None + and xparam.default is not xparam.empty + else value + ) + return lambda: alphaconf.get(key, ktype) + + +def inject(name: str, factory: Union[None, str, Callable[[], Any]]): + """Inject an argument to a function from a factory or alphaconf""" + + def do_inject(func): + f = ParamDefaultsFunction.wrap(func) + if isinstance(factory, str) or factory is None: + b = getter(factory or name, param=f.signature.parameters[name]) + else: + b = factory + f.bind(name, b) + return f + + return do_inject + + +def inject_auto(*, prefix: str = "", ignore: set = set()): + """Inject automatically all paramters""" + if prefix and not prefix.endswith("."): + prefix += "." + + def do_inject(func): + f = ParamDefaultsFunction.wrap(func) + for name, param in f.signature.parameters.items(): + if name in ignore: + continue + f.bind(name, getter(prefix + name, param=param)) + return f + + return do_inject diff --git a/alphaconf/internal/configuration.py b/alphaconf/internal/configuration.py index 8da1d1e..083effd 100644 --- a/alphaconf/internal/configuration.py +++ b/alphaconf/internal/configuration.py @@ -1,6 +1,5 @@ import copy import os -import typing import warnings from enum import Enum from typing import ( @@ -19,7 +18,7 @@ from omegaconf import Container, DictConfig, OmegaConf -from .type_resolvers import convert_to_type, pydantic +from .type_resolvers import convert_to_type, pydantic, type_from_annotation T = TypeVar('T') @@ -92,14 +91,14 @@ def get(self, key: Union[str, Type], type=None, *, default=raise_on_missing): ) if value is raise_on_missing: if default is raise_on_missing: - raise ValueError(f"No value for: {key}") + raise KeyError(f"No value for: {key}") return default # check the returned type and convert when necessary if type is not None and isinstance(value, type): return value if isinstance(value, Container): value = OmegaConf.to_object(value) - if type is not None and default is not None: + if type is not None and value is not default: value = convert_to_type(value, type) return value @@ -110,12 +109,12 @@ def __get_type(self, key: Type, *, default=raise_on_missing): key_str = self.__type_path.get(key) if key_str is None: if default is raise_on_missing: - raise ValueError(f"Key not found for type {key}") + raise KeyError(f"Key not found for type {key}") return default try: value = self.get(key_str, key) self.__type_value = value - except ValueError: + except KeyError: if default is raise_on_missing: raise value = default @@ -130,7 +129,7 @@ def setup_configuration( conf: Union[DictConfig, dict, Any], helpers: Dict[str, str] = {}, *, - path: str = "", + prefix: str = "", ): """Add a default configuration @@ -146,10 +145,10 @@ def setup_configuration( conf_type = None if conf_type: # if already registered, set path to None - self.__type_path[conf_type] = None if conf_type in self.__type_path else path + self.__type_path[conf_type] = None if conf_type in self.__type_path else prefix self.__type_value.pop(conf_type, None) - if path and not path.endswith('.'): - path += "." + if prefix and not prefix.endswith('.'): + prefix += "." if isinstance(conf, str): warnings.warn("provide a dict directly", DeprecationWarning) created_config = OmegaConf.create(conf) @@ -157,16 +156,16 @@ def setup_configuration( raise ValueError("The config is not a dict") conf = created_config if isinstance(conf, DictConfig): - config = self.__prepare_dictconfig(conf, path=path) + config = self.__prepare_dictconfig(conf, path=prefix) else: - created_config = self.__prepare_config(conf, path=path) + created_config = self.__prepare_config(conf, path=prefix) if not isinstance(created_config, DictConfig): - raise ValueError("Failed to convert to a DictConfig") + raise TypeError("Failed to convert to a DictConfig") config = created_config - # add path and merge - if path: - config = self.__add_path(config, path.rstrip(".")) - helpers = {path + k: v for k, v in helpers.items()} + # add prefix and merge + if prefix: + config = self.__add_prefix(config, prefix.rstrip(".")) + helpers = {prefix + k: v for k, v in helpers.items()} self._merge([config]) # helpers self.helpers.update(**helpers) @@ -221,12 +220,12 @@ def __prepare_dictconfig( sub_configs = [] for k, v in obj.items_ex(resolve=False): if not isinstance(k, str): - raise ValueError("Expecting only str instances in dict") + raise TypeError("Expecting only str instances in dict") if recursive: v = self.__prepare_config(v, path + k + ".") if '.' in k: obj.pop(k) - sub_configs.append(self.__add_path(v, k)) + sub_configs.append(self.__add_prefix(v, k)) if sub_configs: obj = cast(DictConfig, OmegaConf.unsafe_merge(obj, *sub_configs)) return obj @@ -252,9 +251,6 @@ def __prepare_pydantic(self, obj, path): # pydantic instance, prepare helpers self.__prepare_pydantic(type(obj), path) return obj.model_dump(mode="json") - # parse typing recursively for documentation - for t in typing.get_args(obj): - self.__prepare_pydantic(t, path) # check if not a type if not isinstance(obj, type): return obj @@ -279,13 +275,14 @@ def __prepare_pydantic(self, obj, path): from alphaconf import SECRET_MASKS SECRET_MASKS.append(lambda s: s == path) - elif check_type and field.annotation: - self.__prepare_pydantic(field.annotation, path + k + ".") + elif check_type: + for ftype in type_from_annotation(field.annotation): + self.__prepare_pydantic(ftype, path + k + ".") return defaults return None @staticmethod - def __add_path(config: Any, path: str) -> DictConfig: - for part in reversed(path.split(".")): + def __add_prefix(config: Any, prefix: str) -> DictConfig: + for part in reversed(prefix.split(".")): config = OmegaConf.create({part: config}) return config diff --git a/alphaconf/internal/type_resolvers.py b/alphaconf/internal/type_resolvers.py index be49a47..d63e052 100644 --- a/alphaconf/internal/type_resolvers.py +++ b/alphaconf/internal/type_resolvers.py @@ -1,4 +1,5 @@ import datetime +import typing from pathlib import Path from omegaconf import OmegaConf @@ -16,11 +17,7 @@ """ -def read_text(value): - return Path(value).expanduser().read_text() - - -def parse_bool(value) -> bool: +def _parse_bool(value) -> bool: if isinstance(value, str): value = value.strip().lower() if value in ('no', 'false', 'n', 'f', 'off', 'none', 'null', 'undefined', '0'): @@ -29,14 +26,14 @@ def parse_bool(value) -> bool: TYPE_CONVERTER = { - bool: parse_bool, + bool: _parse_bool, datetime.datetime: datetime.datetime.fromisoformat, datetime.date: lambda s: datetime.datetime.strptime(s, '%Y-%m-%d').date(), datetime.time: datetime.time.fromisoformat, - Path: lambda s: Path(s).expanduser(), + Path: lambda s: Path(str(s)).expanduser(), str: lambda v: str(v), - 'read_text': read_text, - 'read_strip': lambda s: read_text(s).strip(), + 'read_text': lambda s: Path(s).expanduser().read_text(), + 'read_strip': lambda s: Path(s).expanduser().read_text().strip(), 'read_bytes': lambda s: Path(s).expanduser().read_bytes(), } @@ -65,3 +62,12 @@ def convert_to_type(value, type): if pydantic: return pydantic.TypeAdapter(type).validate_python(value) return type(value) + + +def type_from_annotation(annotation) -> typing.Generator[type, None, None]: + """Given an annotation (optional), figure out the types""" + if isinstance(annotation, type) and annotation is not type(None): + yield annotation + else: + for t in typing.get_args(annotation): + yield from type_from_annotation(t) diff --git a/example-typed.py b/example-typed.py index ece0717..fc2b1bd 100755 --- a/example-typed.py +++ b/example-typed.py @@ -20,7 +20,7 @@ class MyConfiguration(BaseModel): connection: Optional[Conn] = None -alphaconf.setup_configuration(MyConfiguration, path="c") +alphaconf.setup_configuration(MyConfiguration, prefix="c") def main(): diff --git a/pyproject.toml b/pyproject.toml index af22d06..b9a7e56 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,13 +23,13 @@ classifiers = [ ] dependencies = [ "omegaconf>=2", - "pydantic>=2", ] [project.optional-dependencies] color = ["colorama"] dotenv = ["python-dotenv"] invoke = ["invoke"] +pydantic = ["pydantic>=2"] toml = ["toml"] [project.urls] diff --git a/tests/test_alphaconf.py b/tests/test_alphaconf.py index 3544057..add95b7 100644 --- a/tests/test_alphaconf.py +++ b/tests/test_alphaconf.py @@ -90,7 +90,7 @@ def test_setup_configuration(): def test_setup_configuration_invalid(): - with pytest.raises(ValueError): + with pytest.raises(TypeError): # invalid configuration (must be non-empty) alphaconf.setup_configuration(None) @@ -132,8 +132,8 @@ def test_app_environ(application): ) application.setup_configuration(load_dotenv=False, env_prefixes=True) config = application.configuration - with pytest.raises(ValueError): - # XXX should not be loaded + with pytest.raises(KeyError): + # prefix with underscore only should be loaded config.get('xxx') assert config.get('testmyenv.x') == 'overwrite' assert config.get('testmyenv.y') == 'new' diff --git a/tests/test_configuration.py b/tests/test_configuration.py index b9386f0..05fa83b 100644 --- a/tests/test_configuration.py +++ b/tests/test_configuration.py @@ -48,6 +48,7 @@ def test_cast(config): # cast bool into int assert config.get('b') is True assert config.get('b', int) == 1 + assert config.get('b', int, default=None) == 1 # cast Path assert isinstance(config.get('home', Path), Path) @@ -71,7 +72,7 @@ def test_select_empty(config): def test_select_required(config): assert config.get('z', default=None) is None - with pytest.raises(ValueError): + with pytest.raises(KeyError): print(config.get('z')) assert config.get('z', default='a') == 'a' @@ -80,7 +81,7 @@ def test_select_required_incomplete(config_req): # when we have a default, return it assert config_req.get('req', default='def') == 'def' # when required, raise missing - with pytest.raises(ValueError): + with pytest.raises(KeyError): print(config_req.get('req')) @@ -132,5 +133,5 @@ def test_config_setup_dots(config): def test_config_setup_path(config): - config.setup_configuration({'test': 954}, path='a.b') + config.setup_configuration({'test': 954}, prefix='a.b') assert config.get('a.b.test') == 954 diff --git a/tests/test_configuration_typed.py b/tests/test_configuration_typed.py index 3f1808d..5b55097 100644 --- a/tests/test_configuration_typed.py +++ b/tests/test_configuration_typed.py @@ -95,6 +95,6 @@ def test_set_person(config_typed): def test_set_person_type(config_typed): - config_typed.setup_configuration(Person(first_name='A', last_name='T'), path='x_person') + config_typed.setup_configuration(Person(first_name='A', last_name='T'), prefix='x_person') person = config_typed.get(Person) assert person.full_name == 'A T' diff --git a/tests/test_inject.py b/tests/test_inject.py new file mode 100644 index 0000000..05a73f0 --- /dev/null +++ b/tests/test_inject.py @@ -0,0 +1,81 @@ +import string +from typing import Optional + +import pytest + +import alphaconf +import alphaconf.inject as inj + + +@pytest.fixture(scope="function") +def c(): + alphaconf.setup_configuration(dict(zip(string.ascii_letters, range(1, 11)))) + alphaconf.set_application(app := alphaconf.Application()) + return app.configuration + + +def mytuple(a: int, b=1, *, c, d=1, zz: int = 1): + return (a, b, c, d, zz) + + +@pytest.fixture(scope="function") +def mytupledef(): + return inj.inject("a", lambda: 1)(inj.inject("c", lambda: 1)(mytuple)) + + +def test_inject(c, mytupledef): + assert mytuple(0, c=2) == (0, 1, 2, 1, 1) + assert mytupledef() == (1, 1, 1, 1, 1) + assert inj.inject("c", lambda: 5)(mytuple)(0) == (0, 1, 5, 1, 1) + assert inj.inject("b", lambda: 5)(mytupledef)() == (1, 5, 1, 1, 1) + + +def test_inject_name(c, mytupledef): + assert inj.inject('a', 'g')(mytuple)(c=0) == (7, 1, 0, 1, 1) + + +def test_inject_auto_lambda(c): + assert inj.inject_auto()(lambda a: a + 1)() == 2 + assert inj.inject_auto()(lambda c: c + 1)() == 4 + + +def test_inject_auto(c): + assert inj.inject_auto()(mytuple)() == (1, 2, 3, 4, 1) + + +def test_inject_auto_ignore(c): + assert inj.inject_auto(ignore={'b'})(mytuple)() == (1, 1, 3, 4, 1) + + +def test_inject_auto_missing(): + with pytest.raises(KeyError, match=": a"): + inj.inject_auto()(mytuple)() + + +def test_inject_auto_prefix(): + def f1(name): + return name + + alphaconf.setup_configuration({"mytest.name": "ok"}) + assert inj.inject_auto(prefix="mytest")(f1)() == "ok" + + +def test_inject_type_def(mytupledef): + with pytest.raises(KeyError): + inj.inject("a", "nothing")(mytupledef)() + + +def test_inject_type_cast(c): + def f1(zz: str): + return zz + + def f2(zz: str = "ok"): + return zz + + def f3(zz: Optional[str] = None): + return zz + + assert inj.inject('zz', 'g')(f1)() == "7" + assert inj.inject('zz', 'nothing')(f2)() == "ok" + assert inj.inject('zz', 'nothing')(f3)() is None + assert inj.inject('zz', 'g')(f3)() == "7"