263 lines
7.2 KiB
Python
263 lines
7.2 KiB
Python
from __future__ import unicode_literals
|
|
|
|
import logging
|
|
import re
|
|
import socket
|
|
|
|
from mopidy.utils import path
|
|
from mopidy.config import validators
|
|
|
|
|
|
def decode(value):
|
|
if isinstance(value, unicode):
|
|
return value
|
|
# TODO: only unescape \n \t and \\?
|
|
return value.decode('string-escape').decode('utf-8')
|
|
|
|
|
|
def encode(value):
|
|
if not isinstance(value, unicode):
|
|
return value
|
|
for char in ('\\', '\n', '\t'): # TODO: more escapes?
|
|
value = value.replace(char, char.encode('unicode-escape'))
|
|
return value.encode('utf-8')
|
|
|
|
|
|
class ExpandedPath(bytes):
|
|
def __new__(self, value):
|
|
expanded = path.expand_path(value)
|
|
return super(ExpandedPath, self).__new__(self, expanded)
|
|
|
|
def __init__(self, value):
|
|
self.original = value
|
|
|
|
|
|
class ConfigValue(object):
|
|
"""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.
|
|
"""
|
|
|
|
def deserialize(self, value):
|
|
"""Cast raw string to appropriate type."""
|
|
return value
|
|
|
|
def serialize(self, value):
|
|
"""Convert value back to string for saving."""
|
|
return bytes(value)
|
|
|
|
def format(self, value):
|
|
"""Format value for display."""
|
|
return self.serialize(value)
|
|
|
|
|
|
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):
|
|
self._required = not optional
|
|
self._choices = choices
|
|
|
|
def deserialize(self, value):
|
|
value = decode(value).strip()
|
|
validators.validate_required(value, self._required)
|
|
validators.validate_choice(value, self._choices)
|
|
if not value:
|
|
return None
|
|
return value
|
|
|
|
def serialize(self, value):
|
|
if value is None:
|
|
return b''
|
|
return encode(value)
|
|
|
|
|
|
class Secret(ConfigValue):
|
|
"""Secret value.
|
|
|
|
Should be used for passwords, auth tokens etc. Deserializing will not
|
|
convert to unicode. Will mask value when being displayed.
|
|
"""
|
|
def __init__(self, optional=False, choices=None):
|
|
self._required = not optional
|
|
|
|
def deserialize(self, value):
|
|
value = value.strip()
|
|
validators.validate_required(value, self._required)
|
|
if not value:
|
|
return None
|
|
return value
|
|
|
|
def serialize(self, value):
|
|
if value is None:
|
|
return b''
|
|
return value
|
|
|
|
def format(self, value):
|
|
if value is None:
|
|
return b''
|
|
return b'********'
|
|
|
|
|
|
class Integer(ConfigValue):
|
|
"""Integer value."""
|
|
|
|
def __init__(self, minimum=None, maximum=None, choices=None):
|
|
self._minimum = minimum
|
|
self._maximum = maximum
|
|
self._choices = choices
|
|
|
|
def deserialize(self, value):
|
|
value = int(value)
|
|
validators.validate_choice(value, self._choices)
|
|
validators.validate_minimum(value, self._minimum)
|
|
validators.validate_maximum(value, self._maximum)
|
|
return value
|
|
|
|
|
|
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 deserialize(self, value):
|
|
if value.lower() in self.true_values:
|
|
return True
|
|
elif value.lower() in self.false_values:
|
|
return False
|
|
raise ValueError('invalid value for boolean: %r' % value)
|
|
|
|
def serialize(self, value):
|
|
if value:
|
|
return 'true'
|
|
else:
|
|
return 'false'
|
|
|
|
|
|
class List(ConfigValue):
|
|
"""List value.
|
|
|
|
Supports elements split by commas or newlines. Newlines take presedence and
|
|
empty list items will be filtered out.
|
|
"""
|
|
def __init__(self, optional=False):
|
|
self._required = not optional
|
|
|
|
def deserialize(self, value):
|
|
if b'\n' in value:
|
|
values = re.split(r'\s*\n\s*', value)
|
|
else:
|
|
values = re.split(r'\s*,\s*', value)
|
|
values = (decode(v).strip() for v in values)
|
|
values = filter(None, values)
|
|
validators.validate_required(values, self._required)
|
|
return tuple(values)
|
|
|
|
def serialize(self, value):
|
|
return b'\n ' + b'\n '.join(encode(v) for v in value if v)
|
|
|
|
|
|
class LogLevel(ConfigValue):
|
|
"""Log level value.
|
|
|
|
Expects one of ``critical``, ``error``, ``warning``, ``info``, ``debug``
|
|
with any casing.
|
|
"""
|
|
levels = {
|
|
'critical': logging.CRITICAL,
|
|
'error': logging.ERROR,
|
|
'warning': logging.WARNING,
|
|
'info': logging.INFO,
|
|
'debug': logging.DEBUG,
|
|
}
|
|
|
|
def deserialize(self, value):
|
|
validators.validate_choice(value.lower(), self.levels.keys())
|
|
return self.levels.get(value.lower())
|
|
|
|
def serialize(self, value):
|
|
return dict((v, k) for k, v in self.levels.items()).get(value)
|
|
|
|
|
|
class Hostname(ConfigValue):
|
|
"""Network hostname value."""
|
|
|
|
def __init__(self, optional=False):
|
|
self._required = not optional
|
|
|
|
def deserialize(self, value):
|
|
validators.validate_required(value, self._required)
|
|
if not value.strip():
|
|
return None
|
|
try:
|
|
socket.getaddrinfo(value, None)
|
|
except socket.error:
|
|
raise ValueError('must be a resolveable hostname or valid IP')
|
|
return value
|
|
|
|
|
|
class Port(Integer):
|
|
"""Network port value.
|
|
|
|
Expects integer in the range 0-65535, zero tells the kernel to simply
|
|
allocate a port for us.
|
|
"""
|
|
# TODO: consider probing if port is free or not?
|
|
def __init__(self, choices=None):
|
|
super(Port, self).__init__(minimum=0, maximum=2**16-1, choices=choices)
|
|
|
|
|
|
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
|
|
|
|
Supported kwargs: ``optional``, ``choices``, and ``secret``
|
|
"""
|
|
def __init__(self, optional=False, choices=None):
|
|
self._required = not optional
|
|
self._choices = choices
|
|
|
|
def deserialize(self, value):
|
|
value = value.strip()
|
|
validators.validate_required(value, self._required)
|
|
validators.validate_choice(value, self._choices)
|
|
if not value:
|
|
return None
|
|
return ExpandedPath(value)
|
|
|
|
def serialize(self, value):
|
|
if isinstance(value, ExpandedPath):
|
|
return value.original
|
|
return value
|