Source code for mopidy.config.types

import logging
import re
import socket

from mopidy.config import validators
from mopidy.internal import log, path


def decode(value):
    if isinstance(value, bytes):
        value = value.decode(errors="surrogateescape")

    for char in ("\\", "\n", "\t"):
        value = value.replace(
            char.encode(encoding="unicode-escape").decode(), char
        )

    return value


def encode(value):
    if isinstance(value, bytes):
        value = value.decode(errors="surrogateescape")

    for char in ("\\", "\n", "\t"):
        value = value.replace(
            char, char.encode(encoding="unicode-escape").decode()
        )

    return value


class DeprecatedValue:
    pass


class _TransformedValue(str):
    def __new__(cls, original, transformed):
        return super().__new__(cls, transformed)

    def __init__(self, original, transformed):
        self.original = original


[docs]class ConfigValue: """Represents a config key's value and how to handle it. Normally you will only be interacting with sub-classes for config values that encode either deserialization behavior and/or validation. Each config value should be used for the following actions: 1. Deserializing from a raw string and validating, raising ValueError on failure. 2. Serializing a value back to a string that can be stored in a config. 3. Formatting a value to a printable form (useful for masking secrets). :class:`None` values should not be deserialized, serialized or formatted, the code interacting with the config should simply skip None config values. """
[docs] def deserialize(self, value): """Cast raw string to appropriate type.""" return decode(value)
[docs] def serialize(self, value, display=False): """Convert value back to string for saving.""" if value is None: return "" return str(value)
[docs]class Deprecated(ConfigValue): """Deprecated value. Used for ignoring old config values that are no longer in use, but should not cause the config parser to crash. """
[docs] def deserialize(self, value): return DeprecatedValue()
[docs] def serialize(self, value, display=False): return DeprecatedValue()
[docs]class String(ConfigValue): """String value. Is decoded as utf-8 and \\n \\t escapes should work and be preserved. """ def __init__(self, optional=False, choices=None, transformer=None): self._required = not optional self._choices = choices self._transformer = transformer
[docs] def deserialize(self, value): value = decode(value).strip() validators.validate_required(value, self._required) if not value: return None # This is necessary for backwards-compatibility, in case subclasses # aren't calling their parent constructor. transformer = getattr(self, "_transformer", None) if transformer: transformed_value = transformer(value) value = _TransformedValue(value, transformed_value) validators.validate_choice(value, self._choices) return value
[docs] def serialize(self, value, display=False): if value is None: return "" if isinstance(value, _TransformedValue): value = value.original return encode(value)
[docs]class Secret(String): """Secret string value. Is decoded as utf-8 and \\n \\t escapes should work and be preserved. Should be used for passwords, auth tokens etc. Will mask value when being displayed. """ def __init__(self, optional=False, choices=None, transformer=None): super().__init__( optional=optional, choices=None, # Choices doesn't make sense for secrets transformer=transformer, )
[docs] def serialize(self, value, display=False): if value is not None and display: return "********" return super().serialize(value, display)
[docs]class Integer(ConfigValue): """Integer value.""" def __init__( self, minimum=None, maximum=None, choices=None, optional=False ): self._required = not optional self._minimum = minimum self._maximum = maximum self._choices = choices
[docs] def deserialize(self, value): value = decode(value) validators.validate_required(value, self._required) if not value: return None value = int(value) validators.validate_choice(value, self._choices) validators.validate_minimum(value, self._minimum) validators.validate_maximum(value, self._maximum) return value
[docs]class Float(ConfigValue): """Float value.""" def __init__(self, minimum=None, maximum=None, optional=False): self._required = not optional self._minimum = minimum self._maximum = maximum
[docs] def deserialize(self, value): value = decode(value) validators.validate_required(value, self._required) if not value: return None value = float(value) validators.validate_minimum(value, self._minimum) validators.validate_maximum(value, self._maximum) return value
[docs]class Boolean(ConfigValue): """Boolean value. Accepts ``1``, ``yes``, ``true``, and ``on`` with any casing as :class:`True`. Accepts ``0``, ``no``, ``false``, and ``off`` with any casing as :class:`False`. """ true_values = ("1", "yes", "true", "on") false_values = ("0", "no", "false", "off") def __init__(self, optional=False): self._required = not optional
[docs] def deserialize(self, value): value = decode(value) validators.validate_required(value, self._required) if not value: return None if value.lower() in self.true_values: return True elif value.lower() in self.false_values: return False raise ValueError(f"invalid value for boolean: {value!r}")
[docs] def serialize(self, value, display=False): if value is True: return "true" elif value in (False, None): return "false" else: raise ValueError(f"{value!r} is not a boolean")
[docs]class Pair(ConfigValue): """Pair value The value is expected to be a pair of elements, separated by a specified delimiter. Values can optionally not be a pair, in which case the whole input is provided for both sides of the value. """ def __init__( self, optional=False, optional_pair=False, separator="|", subtypes=None ): self._required = not optional self._optional_pair = optional_pair self._separator = separator if subtypes: self._subtypes = subtypes else: self._subtypes = (String(), String())
[docs] def deserialize(self, value): raw_value = decode(value).strip() validators.validate_required(raw_value, self._required) if not raw_value: return None if self._separator in raw_value: values = raw_value.split(self._separator, 1) elif self._optional_pair: values = (raw_value, raw_value) else: raise ValueError( f"Config value must include {self._separator!r} separator: {raw_value}" ) return ( self._subtypes[0].deserialize(encode(values[0])), self._subtypes[1].deserialize(encode(values[1])), )
[docs] def serialize(self, value, display=False): serialized_first_value = self._subtypes[0].serialize( value[0], display=display ) serialized_second_value = self._subtypes[1].serialize( value[1], display=display ) if ( not display and self._optional_pair and serialized_first_value == serialized_second_value ): return serialized_first_value else: return "{0}{1}{2}".format( serialized_first_value, self._separator, serialized_second_value, )
[docs]class List(ConfigValue): """List value. Supports elements split by commas or newlines. Newlines take precedence and empty list items will be filtered out. Enforcing unique entries in the list will result in a set data structure being used. This does not preserve ordering, which could result in the serialized output being unstable. """ def __init__(self, optional=False, unique=False, subtype=None): self._required = not optional self._unique = unique self._subtype = subtype if subtype else String()
[docs] def deserialize(self, value): value = decode(value) if "\n" in value: values = re.split(r"\s*\n\s*", value) else: values = re.split(r"\s*,\s*", value) # This is necessary for backwards-compatibility, in case subclasses # aren't calling their parent constructor. subtype = getattr(self, "_subtype", String()) values_iter = ( subtype.deserialize(v.strip()) for v in values if v.strip() ) if self._unique: values = frozenset(values_iter) else: values = tuple(values_iter) validators.validate_required(values, self._required) return values
[docs] def serialize(self, value, display=False): if not value: return "" # This is necessary for backwards-compatibility, in case subclasses # aren't calling their parent constructor. subtype = getattr(self, "_subtype", String()) serialized_values = [] for item in value: serialized_value = subtype.serialize(item, display=display) if serialized_value: serialized_values.append(serialized_value) return "\n " + "\n ".join(serialized_values)
[docs]class LogColor(ConfigValue):
[docs] def deserialize(self, value): value = decode(value) validators.validate_choice(value.lower(), log.COLORS) return value.lower()
[docs] def serialize(self, value, display=False): if value.lower() in log.COLORS: return encode(value.lower()) return ""
[docs]class LogLevel(ConfigValue): """Log level value. Expects one of ``critical``, ``error``, ``warning``, ``info``, ``debug``, ``trace``, or ``all``, with any casing. """ levels = { "critical": logging.CRITICAL, "error": logging.ERROR, "warning": logging.WARNING, "info": logging.INFO, "debug": logging.DEBUG, "trace": log.TRACE_LOG_LEVEL, "all": logging.NOTSET, }
[docs] def deserialize(self, value): value = decode(value) validators.validate_choice(value.lower(), self.levels.keys()) return self.levels.get(value.lower())
[docs] def serialize(self, value, display=False): lookup = {v: k for k, v in self.levels.items()} if value in lookup: return encode(lookup[value]) return ""
[docs]class Hostname(ConfigValue): """Network hostname value.""" def __init__(self, optional=False): self._required = not optional
[docs] def deserialize(self, value, display=False): value = decode(value).strip() validators.validate_required(value, self._required) if not value: return None socket_path = path.get_unix_socket_path(value) if socket_path is not None: path_str = Path(not self._required).deserialize(socket_path) return f"unix:{path_str}" try: socket.getaddrinfo(value, None) except OSError: raise ValueError("must be a resolveable hostname or valid IP") return value
[docs]class Port(Integer): """Network port value. Expects integer in the range 0-65535, zero tells the kernel to simply allocate a port for us. """ def __init__(self, choices=None, optional=False): super().__init__( minimum=0, maximum=2**16 - 1, choices=choices, optional=optional )
# Keep this for backwards compatibility class _ExpandedPath(_TransformedValue): pass
[docs]class Path(ConfigValue): """File system path. The following expansions of the path will be done: - ``~`` to the current user's home directory - ``$XDG_CACHE_DIR`` according to the XDG spec - ``$XDG_CONFIG_DIR`` according to the XDG spec - ``$XDG_DATA_DIR`` according to the XDG spec - ``$XDG_MUSIC_DIR`` according to the XDG spec """ def __init__(self, optional=False): self._required = not optional
[docs] def deserialize(self, value): value = decode(value).strip() expanded = path.expand_path(value) validators.validate_required(value, self._required) validators.validate_required(expanded, self._required) if not value or expanded is None: return None return _ExpandedPath(value, expanded)
[docs] def serialize(self, value, display=False): if value is None: return "" if isinstance(value, _ExpandedPath): value = value.original if isinstance(value, bytes): value = value.decode(errors="surrogateescape") return value