Source code for mtap._config

# Copyright 2019 Regents of the University of Minnesota.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""MTAP SDK configuration."""
import logging
import os
import threading
from pathlib import Path
from typing import Any, MutableMapping, Iterator, Union

from mtap.utilities import mtap_home

logger = logging.getLogger('mtap.config')


def _collapse(d, path, v):
    d[path] = v
    try:
        p = ''
        if path is not None:
            p = path + '.'
        for k, v in v.items():
            _collapse(d, p + k, v)
        return d
    except AttributeError:
        pass
    except TypeError:
        raise ValueError('Failed to load configuration')
    return d


def _load_config(f):
    from yaml import load
    try:
        from yaml import CLoader as Loader
    except ImportError:
        from yaml import Loader
    config = load(f, Loader=Loader)
    if not isinstance(config, dict):
        raise TypeError("Failed to load configuration from file: " + str(f))
    return _collapse({}, None, config)


def _load_default_config():
    potential_paths = []
    try:
        cnf = os.getenv('MTAP_CONFIG')
        potential_paths.append(Path(cnf))
    except TypeError:
        pass
    locations = [Path.cwd(), mtap_home(), Path('/etc/mtap/')]
    potential_paths += [location / 'mtapConfig.yml' for location in locations]
    potential_paths.append(Path(__file__).parent / 'defaultConfig.yml')
    for config_path in potential_paths:
        try:
            with config_path.open('rb') as f:
                logger.debug('Loading MTAP config from "{}"'.format(config_path))
                return _load_config(f)
        except FileNotFoundError:
            pass
    raise ValueError('Failed to load configuration file from')


[docs]class Config(MutableMapping[str, Any]): """The MTAP configuration dictionary. By default configuration is loaded from one of a number of locations in the following priority: - A file at the path of the '--config' parameter passed into main methods. - A file at the path of the 'MTAP_CONFIG' environment variable - $PWD/mtapConfig.yml - $HOME/.mtap/mtapConfig.yml' - /etc/mtap/mtapConfig.yml MTAP components will use a global shared configuration object, by entering the context of a config object using "with", all of the MTAP functions called on that thread will make use of that config object. Examples: >>> with mtap.Config() as config: >>> config['key'] = 'value' >>> # other MTAP methods in this >>> # block will use the updated config object. """ _lock = threading.RLock() _global_instance = None _context = threading.local() _context.config = None def __new__(cls, *args): if cls._global_instance is None: with cls._lock: if cls._global_instance is None: cls._global_instance = object.__new__(cls) cls._global_instance._config = {} if len(args) == 0: cls._global_instance._load_default_config() else: cls._global_instance._config.update(*args) try: inst = cls._context.config except AttributeError: cls._context.config = None inst = None if inst is not None: return inst inst = object.__new__(cls) inst._config = {} inst.update(cls._global_instance) return inst def __enter__(self) -> 'Config': self.enter_context() return self def enter_context(self): with self._lock: if hasattr(self._context, "config") and self._context.config is not None: raise ValueError("Already in a configuration context.") self._context.config = self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def close(self): self._context.config = None def __reduce__(self): return Config, (dict(self), ) def _load_default_config(self): self.update(_load_default_config())
[docs] def update_from_yaml(self, path: Union[Path, str]): """Updates the configuration by loading and collapsing all of the structures in a yaml file. Parameters: path: The path to the yaml file to load. """ if isinstance(path, str): path = Path(path) with path.open('rb') as f: self.update(_load_config(f))
def __setitem__(self, k: str, v: Any) -> None: self._config[k] = v def __delitem__(self, v: str) -> None: del self._config[v] def __getitem__(self, k: str) -> Any: return self._config[k] def __len__(self) -> int: return len(self._config) def __iter__(self) -> Iterator[str]: return iter(self._config)