dvc/dvc/config.py

"""
DVC config objects.
"""
import os
import errno
import configobj
from schema import Schema, Optional, And, Use, Regex

import dvc.logger as logger
from dvc.exceptions import DvcException


class ConfigError(DvcException):
    """DVC config exception.

    Args:
        msg (str): error message.
        ex (Exception): optional exception that has caused this error.
    """
    def __init__(self, msg, ex=None):
        super(ConfigError, self).__init__('config file error: {}'.format(msg),
                                          ex)


def supported_cache_type(types):
    """Checks if link type config option has a valid value.

    Args:
        types (list/string): type(s) of links that dvc should try out.
    """
    if isinstance(types, str):
        types = [typ.strip() for typ in types.split(',')]
    for typ in types:
        if typ not in ['reflink', 'hardlink', 'symlink', 'copy']:
            return False
    return True


def supported_loglevel(level):
    """Checks if log level config option has a valid value.

    Args:
        level (str): log level name.
    """
    return level in ['info', 'debug', 'warning', 'error']


def supported_cloud(cloud):
    """Checks if obsoleted cloud option has a valid value.

    Args:
        cloud (str): cloud type name.
    """
    return cloud in ['aws', 'gcp', 'local', '']


def is_bool(val):
    """Checks that value is a boolean.

    Args:
        val (str): string value verify.

    Returns:
        bool: True if value stands for boolean, False otherwise.
    """
    return val.lower() in ['true', 'false']


def to_bool(val):
    """Converts value to boolean.

    Args:
        val (str): string to convert to boolean.

    Returns:
        bool: True if value.lower() == 'true', False otherwise.
    """
    return val.lower() == 'true'


def is_whole(val):
    """Checks that value is a whole integer.

    Args:
        val (str): number string to verify.

    Returns:
        bool: True if val is a whole number, False otherwise.
    """
    return int(val) >= 0


def is_percent(val):
    """Checks that value is a percent.

    Args:
        val (str): number string to verify.

    Returns:
        bool: True if 0<=value<=100, False otherwise.
    """
    return int(val) >= 0 and int(val) <= 100


class Config(object):  # pylint: disable=too-many-instance-attributes
    """Class that manages configuration files for a dvc project.

    Args:
        dvc_dir (str): optional path to `.dvc` directory, that is used to
            access project-specific configs like .dvc/config and
            .dvc/config.local.
        validate (bool): optional flag to tell dvc if it should validate the
            config or just load it as is. 'True' by default.


    Raises:
        ConfigError: thrown when config has an invalid format.
    """
    APPNAME = 'dvc'
    APPAUTHOR = 'iterative'

    # NOTE: used internally in RemoteLOCAL to know config
    # location, that url should resolved relative to.
    PRIVATE_CWD = '_cwd'

    CONFIG = 'config'
    CONFIG_LOCAL = 'config.local'

    SECTION_CORE = 'core'
    SECTION_CORE_LOGLEVEL = 'loglevel'
    SECTION_CORE_LOGLEVEL_SCHEMA = And(Use(str.lower), supported_loglevel)
    SECTION_CORE_REMOTE = 'remote'
    SECTION_CORE_INTERACTIVE_SCHEMA = And(str, is_bool, Use(to_bool))
    SECTION_CORE_INTERACTIVE = 'interactive'
    SECTION_CORE_ANALYTICS = 'analytics'
    SECTION_CORE_ANALYTICS_SCHEMA = And(str, is_bool, Use(to_bool))

    SECTION_CACHE = 'cache'
    SECTION_CACHE_DIR = 'dir'
    SECTION_CACHE_TYPE = 'type'
    SECTION_CACHE_TYPE_SCHEMA = supported_cache_type
    SECTION_CACHE_PROTECTED = 'protected'
    SECTION_CACHE_LOCAL = 'local'
    SECTION_CACHE_S3 = 's3'
    SECTION_CACHE_GS = 'gs'
    SECTION_CACHE_SSH = 'ssh'
    SECTION_CACHE_HDFS = 'hdfs'
    SECTION_CACHE_AZURE = 'azure'
    SECTION_CACHE_SCHEMA = {
        Optional(SECTION_CACHE_LOCAL): str,
        Optional(SECTION_CACHE_S3): str,
        Optional(SECTION_CACHE_GS): str,
        Optional(SECTION_CACHE_HDFS): str,
        Optional(SECTION_CACHE_SSH): str,
        Optional(SECTION_CACHE_AZURE): str,

        Optional(SECTION_CACHE_DIR): str,
        Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA,
        Optional(SECTION_CACHE_PROTECTED,
                 default=False): And(str, is_bool, Use(to_bool)),

        Optional(PRIVATE_CWD): str,
    }

    # backward compatibility
    SECTION_CORE_CLOUD = 'cloud'
    SECTION_CORE_CLOUD_SCHEMA = And(Use(str.lower), supported_cloud)
    SECTION_CORE_STORAGEPATH = 'storagepath'

    SECTION_CORE_SCHEMA = {
        Optional(SECTION_CORE_LOGLEVEL,
                 default='info'): And(str, Use(str.lower),
                                      SECTION_CORE_LOGLEVEL_SCHEMA),
        Optional(SECTION_CORE_REMOTE, default=''): And(str, Use(str.lower)),
        Optional(SECTION_CORE_INTERACTIVE,
                 default=False): SECTION_CORE_INTERACTIVE_SCHEMA,
        Optional(SECTION_CORE_ANALYTICS,
                 default=True): SECTION_CORE_ANALYTICS_SCHEMA,

        # backward compatibility
        Optional(SECTION_CORE_CLOUD, default=''): SECTION_CORE_CLOUD_SCHEMA,
        Optional(SECTION_CORE_STORAGEPATH, default=''): str,
    }

    # backward compatibility
    SECTION_AWS = 'aws'
    SECTION_AWS_STORAGEPATH = 'storagepath'
    SECTION_AWS_CREDENTIALPATH = 'credentialpath'
    SECTION_AWS_ENDPOINT_URL = 'endpointurl'
    SECTION_AWS_REGION = 'region'
    SECTION_AWS_PROFILE = 'profile'
    SECTION_AWS_USE_SSL = 'use_ssl'
    SECTION_AWS_SCHEMA = {
        SECTION_AWS_STORAGEPATH: str,
        Optional(SECTION_AWS_REGION): str,
        Optional(SECTION_AWS_PROFILE): str,
        Optional(SECTION_AWS_CREDENTIALPATH): str,
        Optional(SECTION_AWS_ENDPOINT_URL): str,
        Optional(SECTION_AWS_USE_SSL,
                 default=True): And(str, is_bool, Use(to_bool)),
    }

    # backward compatibility
    SECTION_GCP = 'gcp'
    SECTION_GCP_STORAGEPATH = SECTION_AWS_STORAGEPATH
    SECTION_GCP_PROJECTNAME = 'projectname'
    SECTION_GCP_SCHEMA = {
        SECTION_GCP_STORAGEPATH: str,
        Optional(SECTION_GCP_PROJECTNAME): str,
    }

    # backward compatibility
    SECTION_LOCAL = 'local'
    SECTION_LOCAL_STORAGEPATH = SECTION_AWS_STORAGEPATH
    SECTION_LOCAL_SCHEMA = {
        SECTION_LOCAL_STORAGEPATH: str,
    }

    SECTION_AZURE_CONNECTION_STRING = 'connection_string'

    SECTION_REMOTE_REGEX = r'^\s*remote\s*"(?P<name>.*)"\s*$'
    SECTION_REMOTE_FMT = 'remote "{}"'
    SECTION_REMOTE_URL = 'url'
    SECTION_REMOTE_USER = 'user'
    SECTION_REMOTE_PORT = 'port'
    SECTION_REMOTE_KEY_FILE = 'keyfile'
    SECTION_REMOTE_TIMEOUT = 'timeout'
    SECTION_REMOTE_PASSWORD = 'password'
    SECTION_REMOTE_ASK_PASSWORD = 'ask_password'
    SECTION_REMOTE_SCHEMA = {
        SECTION_REMOTE_URL: str,
        Optional(SECTION_AWS_REGION): str,
        Optional(SECTION_AWS_PROFILE): str,
        Optional(SECTION_AWS_CREDENTIALPATH): str,
        Optional(SECTION_AWS_ENDPOINT_URL): str,
        Optional(SECTION_AWS_USE_SSL,
                 default=True): And(str, is_bool, Use(to_bool)),
        Optional(SECTION_GCP_PROJECTNAME): str,
        Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA,
        Optional(SECTION_CACHE_PROTECTED,
                 default=False): And(str, is_bool, Use(to_bool)),
        Optional(SECTION_REMOTE_USER): str,
        Optional(SECTION_REMOTE_PORT): Use(int),
        Optional(SECTION_REMOTE_KEY_FILE): str,
        Optional(SECTION_REMOTE_TIMEOUT): Use(int),
        Optional(SECTION_REMOTE_PASSWORD): str,
        Optional(SECTION_REMOTE_ASK_PASSWORD): And(str, is_bool, Use(to_bool)),
        Optional(SECTION_AZURE_CONNECTION_STRING): str,

        Optional(PRIVATE_CWD): str,
    }

    SECTION_STATE = 'state'
    SECTION_STATE_ROW_LIMIT = 'row_limit'
    SECTION_STATE_ROW_CLEANUP_QUOTA = 'row_cleanup_quota'
    SECTION_STATE_SCHEMA = {
        Optional(SECTION_STATE_ROW_LIMIT): And(Use(int), is_whole),
        Optional(SECTION_STATE_ROW_CLEANUP_QUOTA): And(Use(int), is_percent),
    }

    SCHEMA = {
        Optional(SECTION_CORE, default={}): SECTION_CORE_SCHEMA,
        Optional(Regex(SECTION_REMOTE_REGEX)): SECTION_REMOTE_SCHEMA,
        Optional(SECTION_CACHE, default={}): SECTION_CACHE_SCHEMA,
        Optional(SECTION_STATE, default={}): SECTION_STATE_SCHEMA,

        # backward compatibility
        Optional(SECTION_AWS, default={}): SECTION_AWS_SCHEMA,
        Optional(SECTION_GCP, default={}): SECTION_GCP_SCHEMA,
        Optional(SECTION_LOCAL, default={}): SECTION_LOCAL_SCHEMA,
    }

    def __init__(self, dvc_dir=None, validate=True):
        self.system_config_file = os.path.join(self.get_system_config_dir(),
                                               self.CONFIG)
        self.global_config_file = os.path.join(self.get_global_config_dir(),
                                               self.CONFIG)

        if dvc_dir is not None:
            self.dvc_dir = os.path.abspath(os.path.realpath(dvc_dir))
            self.config_file = os.path.join(dvc_dir, self.CONFIG)
            self.config_local_file = os.path.join(dvc_dir, self.CONFIG_LOCAL)
        else:
            self.dvc_dir = None
            self.config_file = None
            self.config_local_file = None

        self._system_config = None
        self._global_config = None
        self._project_config = None
        self._local_config = None

        self.config = None

        self.load(validate=validate)

    @staticmethod
    def get_global_config_dir():
        """Returns global config location. E.g. ~/.config/dvc/config.

        Returns:
            str: path to the global config directory.
        """
        from appdirs import user_config_dir
        return user_config_dir(appname=Config.APPNAME,
                               appauthor=Config.APPAUTHOR)

    @staticmethod
    def get_system_config_dir():
        """Returns system config location. E.g. /etc/dvc.conf.

        Returns:
            str: path to the system config directory.
        """
        from appdirs import site_config_dir
        return site_config_dir(appname=Config.APPNAME,
                               appauthor=Config.APPAUTHOR)

    @staticmethod
    def init(dvc_dir):
        """Initializes dvc config.

        Args:
            dvc_dir (str): path to .dvc directory.

        Returns:
            dvc.config.Config: config object.
        """
        config_file = os.path.join(dvc_dir, Config.CONFIG)
        open(config_file, 'w+').close()
        return Config(dvc_dir)

    def _load(self):
        self._system_config = configobj.ConfigObj(self.system_config_file)
        self._global_config = configobj.ConfigObj(self.global_config_file)

        if self.config_file is not None:
            self._project_config = configobj.ConfigObj(self.config_file)
        else:
            self._project_config = configobj.ConfigObj()

        if self.config_local_file is not None:
            self._local_config = configobj.ConfigObj(self.config_local_file)
        else:
            self._local_config = configobj.ConfigObj()

        self.config = None

    def _load_config(self, path):
        config = configobj.ConfigObj(path)
        config = self._lower(config)
        self._resolve_paths(config, path)
        return config

    @staticmethod
    def _resolve_path(path, config_file):
        assert os.path.isabs(config_file)
        config_dir = os.path.dirname(config_file)
        return os.path.abspath(os.path.join(config_dir, path))

    def _resolve_cache_path(self, config, fname):
        cache = config.get(self.SECTION_CACHE)
        if cache is None:
            return

        cache_dir = cache.get(self.SECTION_CACHE_DIR)
        if cache_dir is None:
            return

        cache[self.PRIVATE_CWD] = os.path.dirname(fname)

    def _resolve_paths(self, config, fname):
        self._resolve_cache_path(config, fname)
        for section in config.values():
            if self.SECTION_REMOTE_URL not in section.keys():
                continue

            section[self.PRIVATE_CWD] = os.path.dirname(fname)

    def load(self, validate=True):
        """Loads config from all the config files.

        Args:
            validate (bool): optional flag to tell dvc if it should validate
                the config or just load it as is. 'True' by default.


        Raises:
            dvc.config.ConfigError: thrown if config has invalid format.
        """
        self._load()
        try:
            self.config = self._load_config(self.system_config_file)
            user = self._load_config(self.global_config_file)
            config = self._load_config(self.config_file)
            local = self._load_config(self.config_local_file)

            # NOTE: schema doesn't support ConfigObj.Section validation, so we
            # need to convert our config to dict before passing it to
            for conf in [user, config, local]:
                self.config = self._merge(self.config, conf)

            if validate:
                self.config = Schema(self.SCHEMA).validate(self.config)

            # NOTE: now converting back to ConfigObj
            self.config = configobj.ConfigObj(self.config,
                                              write_empty_values=True)
            self.config.filename = self.config_file
            self._resolve_paths(self.config, self.config_file)
        except Exception as ex:
            raise ConfigError(ex)

    @staticmethod
    def _get_key(conf, name, add=False):
        for k in conf.keys():
            if k.lower() == name.lower():
                return k

        if add:
            conf[name] = {}
            return name

        return None

    def save(self, config=None):
        """Saves config to config files.

        Args:
            config (configobj.ConfigObj): optional config object to save.

        Raises:
            dvc.config.ConfigError: thrown if failed to write config file.
        """
        if config is not None:
            clist = [config]
        else:
            clist = [self._system_config,
                     self._global_config,
                     self._project_config,
                     self._local_config]

        for conf in clist:
            if conf.filename is None:
                continue

            try:
                logger.debug("Writing '{}'.".format(conf.filename))
                dname = os.path.dirname(os.path.abspath(conf.filename))
                try:
                    os.makedirs(dname)
                except OSError as exc:
                    if exc.errno != errno.EEXIST:
                        raise
                conf.write()
            except Exception as exc:
                msg = "failed to write config '{}'".format(conf.filename)
                raise ConfigError(msg, exc)

    @staticmethod
    def unset(config, section, opt=None):
        """Unsets specified option and/or section in the config.

        Args:
            config (configobj.ConfigObj): config to work on.
            section (str): section name.
            opt (str): optional option name.
        """
        if section not in config.keys():
            raise ConfigError("section '{}' doesn't exist".format(section))

        if opt is None:
            del config[section]
            return

        if opt not in config[section].keys():
            raise ConfigError("option '{}.{}' doesn't exist".format(section,
                                                                    opt))
        del config[section][opt]

        if not config[section]:
            del config[section]

    @staticmethod
    def set(config, section, opt, value):
        """Sets specified option in the config.

        Args:
            config (configobj.ConfigObj): config to work on.
            section (str): section name.
            opt (str): option name.
            value: value to set option to.
        """
        if section not in config.keys():
            config[section] = {}

        config[section][opt] = value

    @staticmethod
    def show(config, section, opt):
        """Prints option value from the config.

        Args:
            config (configobj.ConfigObj): config to work on.
            section (str): section name.
            opt (str): option name.
        """
        if section not in config.keys():
            raise ConfigError("section '{}' doesn't exist".format(section))

        if opt not in config[section].keys():
            raise ConfigError("option '{}.{}' doesn't exist".format(section,
                                                                    opt))

        logger.info(config[section][opt])

    @staticmethod
    def _merge(first, second):
        res = {}
        sections = list(first.keys()) + list(second.keys())
        for section in sections:
            first_copy = first.get(section, {}).copy()
            second_copy = second.get(section, {}).copy()
            first_copy.update(second_copy)
            res[section] = first_copy
        return res

    @staticmethod
    def _lower(config):
        new_config = {}
        for s_key, s_value in config.items():
            new_s = {}
            for key, value in s_value.items():
                new_s[key.lower()] = value
            new_config[s_key.lower()] = new_s
        return new_config