Da

✏️ Editing: config_utils.py

import os
from collections import namedtuple
from configparser import NoSectionError, NoOptionError
from urllib.parse import urlparse, urlunparse
from parallels.core import MigrationError
from parallels.core.registry import Registry
from parallels.core.utils.common import format_list
from parallels.core.utils.common.ip import is_ipv6
from parallels.core.utils.entity import Entity
from parallels.core import messages

Auth = namedtuple('Auth', ('username', 'password'))

class HttpConnectionSettings(Entity):
    def __init__(self, url, auth, host=None):
        self._url = url
        self._auth = auth
        self._host = host

@property
    def url(self):
        return self._url

@property
    def auth(self):
        return self._auth

@property
    def protocol(self):
        return urlparse(self.url).scheme

@property
    def port(self):
        return urlparse(self.url).port

@property
    def host(self):
        return self._host

class ConfigSection(object):
    """Dictionary-like interface to configuration section.

Also allows to refer part of keys starting with given prefix.
    """
    def __init__(self, config, section_name, prefix='', optional=False):
        """Class constructor

:type config: configparser.RawConfigParser
        :type section_name: str
        :type prefix: str
        """
        self.config = config
        self.section_name = section_name
        self.prefix = prefix
        self.optional = optional

def __getitem__(self, name):
        try:
            return self.config.get(self.section_name, self.prefix+name)
        except NoSectionError:
            if self.optional:
                return None
            else:
                raise

def get(self, name, default=None):
        return self.get_or_exec(name, lambda: default)

def getboolean(self, name, default=None):
        return self.getboolean_or_exec(name, lambda: default)

def getint(self, name, default=None):
        return self.getint_or_exec(name, lambda: default)

def get_or_exec(self, name, get_default):
        try:
            return self.config.get(self.section_name, self.prefix+str(name))
        except NoOptionError:
            return get_default()
        except NoSectionError:
            if self.optional:
                return get_default()
            else:
                raise

def getboolean_or_exec(self, name, get_default):
        try:
            return self.config.getboolean(self.section_name, self.prefix+str(name))
        except NoOptionError:
            return get_default()
        except NoSectionError:
            if self.optional:
                return get_default()
            else:
                raise

def getint_or_exec(self, name, get_default):
        try:
            return self.config.getint(self.section_name, self.prefix+str(name))
        except NoOptionError:
            return get_default()
        except NoSectionError:
            if self.optional:
                return get_default()
            else:
                raise

def get_password(self, name, default=None):
        env_full_name = ('%s_%s' % (self.section_name, self.prefix + name)).upper().replace('-', '_')
        password = os.getenv(env_full_name)
        if not password:
            env_short_name = (self.prefix + name).upper().replace('-', '_')
            password = os.getenv(env_short_name)
        if not password and self.get(name + '-type') == 'encrypted':
            raise MigrationError(
                messages.UNABLE_DECRYPT_PASSWORD.format(
                    option=self.prefix + name, section=self.section_name, variable=env_full_name
                )
            )
        if not password:
            password = self.get(name)
        if not password:
            password = default
        return password

def keys(self):
        try:
            return [key for key in self.config.options(self.section_name) if key.startswith(self.prefix)]
        except NoSectionError:
            if self.optional:
                return []
            else:
                raise

def items(self):
        try:
            return [(k, v) for (k, v) in self.config.items(self.section_name) if k.startswith(self.prefix)]
        except NoSectionError:
            if self.optional:
                return []
            else:
                raise

def __contains__(self, name):
        try:
            return self.config.has_option(self.section_name, self.prefix+name)
        except NoSectionError:
            if self.optional:
                return False
            else:
                raise

def prefixed(self, prefix):
        return ConfigSection(self.config, self.section_name, self.prefix+prefix, optional=self.optional)

def with_defaults(self, defaults):
        return ConfigSectionWithDefaults(self, defaults)

def __eq__(self, other):
        return type(self) == type(other) and self.items() == other.items()

class ConfigSectionWithDefaults(object):
    """Configuration section with default values.
    """
    def __init__(self, section, defaults):
        self.section = section
        self.defaults = defaults

def __getitem__(self, name):
        return self.section.get_or_exec(name, lambda: self.defaults.get(name))

def get(self, name, default=None):
        return self.get_or_exec(name, lambda: default)

def get_or_exec(self, name, get_default):
        try:
            return self.section.get_or_exec(name, lambda: self.defaults.get(name))
        except ValueError:
            return get_default()

def get_password(self, name, default=None):
        return self.section.get_password(name, self.defaults[name] if name in self.defaults else default)

def keys(self):
        return set(self.section.keys()) | set(self.defaults.keys())

def __contains__(self, name):
        return name in self.section or name in self.defaults

def with_defaults(self, defaults):
        return ConfigSectionWithDefaults(self, defaults)

def get_option(option, default=None, section='GLOBAL'):
    context = Registry.get_instance().get_context()
    if context is not None and context.config is not None and option in context.config.options(section):
        return context.config.get(section, option)
    else:
        return default

def get_boolean_option(option, default, section='GLOBAL'):
    context = Registry.get_instance().get_context()
    if context is not None and context.config is not None and option in context.config.options(section):
        return context.config.getboolean(section, option)
    else:
        return default

def read_auth(section):
    return Auth(section.get('username'), section.get_password('password'))

def read_url(section):
    if 'url' in section:
        url = section['url']

else:
        host = section.get('ip')
        if not host:
            host = section['host']
        protocol = section['protocol']
        port = int(section['port'])
        path = section['path']

if is_ipv6(host):
            host = '[%s]' % host

url = urlunparse((protocol, '%s:%d' % (host, port), path, '', '', ''))

return url

def read_http_connection_settings(section):
    return HttpConnectionSettings(url=read_url(section), auth=read_auth(section), host=section.get('host'))

def get_sections_list(config, section, option):
    """Parse list of sections in specified option of configuration file and check that each section actually exists

Consider we have list of sections in option
    =============================================
    [GLOBAL]
    sources: source1, source2, source3
    =============================================

If you call this function (for with arguments (config, 'GLOBAL', 'sources')), it will:
    1) Return list of sections: ['source1', 'source2', 'source3']
    2) Check that each of these sections exists in configuration file.
    If section does not exist - exception will be raised. For example, if there is no section like:
    =============================================
    [source3]
    options
    =============================================
    exception will be raised.

:rtype: list[str]
    """
    if option in config.options(section):
        list_str = config.get(section, option)
        list_str = list_str.strip()
        if list_str == '':
            return []
        sections = [s.strip() for s in list_str.split(',')]

for section in sections:
            if not config.has_section(section):
                raise MigrationError(
                    messages.SECTION_DOES_NOT_EXIST.format(section=section, option=option)
                )
        return sections
    else:
        return []

class ConfigSelectionOption(object):
    """Description of configuration file option that may have one value from allowed set"""

@property
    def name(self):
        """Return name of the option

:rtype: str
        """
        raise NotImplementedError()

@property
    def allowed_values(self):
        """Return list of allowed values for this option

:rtype: list[str]
        """
        raise NotImplementedError()

@property
    def case_insensitive(self):
        """Return true if option is case insensitive, False otherwise

:rtype: bool
        """
        return True

@property
    def default(self):
        """Return default value for this option, if it was not specified in configuration file

:rtype: str
        """
        raise NotImplementedError()

def read_selection_option(section, selection_option):
    """Read selection option (option that may have one value from allowed set) from configuration file

:type section: parallels.core.utils.config_utils.ConfigSection
    :type selection_option: parallels.core.utils.config_utils.ConfigSelectionOption
    :rtype: str
    """
    def lowercase_if_case_insensitive(v):
        if selection_option.case_insensitive:
            return v.lower()
        else:
            return v

value = lowercase_if_case_insensitive(section.get(selection_option.name, selection_option.default))
    allowed_values = [
        lowercase_if_case_insensitive(allowed_value) for allowed_value in selection_option.allowed_values
    ]
    if value not in allowed_values:
        raise MigrationError(
            messages.INVALID_OPTION_VALUE.format(
                option_name=selection_option.name,
                specified_value=section.get(selection_option.name),
                allowed_values=format_list(allowed_values)
            )
        )

return value

def global_section(config):
    return ConfigSection(config, 'GLOBAL')

📁 Sid Gifari File Manager