mopidy / mopidy
1 2
import logging
2 2
import re
3 2
import socket
4

5 2
from mopidy.config import validators
6 2
from mopidy.internal import log, path
7

8

9 2
def decode(value):
10 2
    if isinstance(value, bytes):
11 2
        value = value.decode(errors="surrogateescape")
12

13 2
    for char in ("\\", "\n", "\t"):
14 2
        value = value.replace(
15
            char.encode(encoding="unicode-escape").decode(), char
16
        )
17

18 2
    return value
19

20

21 2
def encode(value):
22 2
    if isinstance(value, bytes):
23 2
        value = value.decode(errors="surrogateescape")
24

25 2
    for char in ("\\", "\n", "\t"):
26 2
        value = value.replace(
27
            char, char.encode(encoding="unicode-escape").decode()
28
        )
29

30 2
    return value
31

32

33 2
class DeprecatedValue:
34 2
    pass
35

36

37 2
class ConfigValue:
38
    """Represents a config key's value and how to handle it.
39

40
    Normally you will only be interacting with sub-classes for config values
41
    that encode either deserialization behavior and/or validation.
42

43
    Each config value should be used for the following actions:
44

45
    1. Deserializing from a raw string and validating, raising ValueError on
46
       failure.
47
    2. Serializing a value back to a string that can be stored in a config.
48
    3. Formatting a value to a printable form (useful for masking secrets).
49

50
    :class:`None` values should not be deserialized, serialized or formatted,
51
    the code interacting with the config should simply skip None config values.
52
    """
53

54 2
    def deserialize(self, value):
55
        """Cast raw string to appropriate type."""
56 2
        return decode(value)
57

58 2
    def serialize(self, value, display=False):
59
        """Convert value back to string for saving."""
60 2
        if value is None:
61 2
            return ""
62 2
        return str(value)
63

64

65 2
class Deprecated(ConfigValue):
66
    """Deprecated value.
67

68
    Used for ignoring old config values that are no longer in use, but should
69
    not cause the config parser to crash.
70
    """
71

72 2
    def deserialize(self, value):
73 2
        return DeprecatedValue()
74

75 2
    def serialize(self, value, display=False):
76 2
        return DeprecatedValue()
77

78

79 2
class String(ConfigValue):
80
    """String value.
81

82
    Is decoded as utf-8 and \\n \\t escapes should work and be preserved.
83
    """
84

85 2
    def __init__(self, optional=False, choices=None):
86 2
        self._required = not optional
87 2
        self._choices = choices
88

89 2
    def deserialize(self, value):
90 2
        value = decode(value).strip()
91 2
        validators.validate_required(value, self._required)
92 2
        if not value:
93 2
            return None
94 2
        validators.validate_choice(value, self._choices)
95 2
        return value
96

97 2
    def serialize(self, value, display=False):
98 2
        if value is None:
99 2
            return ""
100 2
        return encode(value)
101

102

103 2
class Secret(String):
104
    """Secret string value.
105

106
    Is decoded as utf-8 and \\n \\t escapes should work and be preserved.
107

108
    Should be used for passwords, auth tokens etc. Will mask value when being
109
    displayed.
110
    """
111

112 2
    def __init__(self, optional=False, choices=None):
113 2
        self._required = not optional
114 2
        self._choices = None  # Choices doesn't make sense for secrets
115

116 2
    def serialize(self, value, display=False):
117 2
        if value is not None and display:
118 2
            return "********"
119 2
        return super().serialize(value, display)
120

121

122 2
class Integer(ConfigValue):
123
    """Integer value."""
124

125 2
    def __init__(
126
        self, minimum=None, maximum=None, choices=None, optional=False
127
    ):
128 2
        self._required = not optional
129 2
        self._minimum = minimum
130 2
        self._maximum = maximum
131 2
        self._choices = choices
132

133 2
    def deserialize(self, value):
134 2
        value = decode(value)
135 2
        validators.validate_required(value, self._required)
136 2
        if not value:
137 2
            return None
138 2
        value = int(value)
139 2
        validators.validate_choice(value, self._choices)
140 2
        validators.validate_minimum(value, self._minimum)
141 2
        validators.validate_maximum(value, self._maximum)
142 2
        return value
143

144

145 2
class Boolean(ConfigValue):
146
    """Boolean value.
147

148
    Accepts ``1``, ``yes``, ``true``, and ``on`` with any casing as
149
    :class:`True`.
150

151
    Accepts ``0``, ``no``, ``false``, and ``off`` with any casing as
152
    :class:`False`.
153
    """
154

155 2
    true_values = ("1", "yes", "true", "on")
156 2
    false_values = ("0", "no", "false", "off")
157

158 2
    def __init__(self, optional=False):
159 2
        self._required = not optional
160

161 2
    def deserialize(self, value):
162 2
        value = decode(value)
163 2
        validators.validate_required(value, self._required)
164 2
        if not value:
165 2
            return None
166 2
        if value.lower() in self.true_values:
167 2
            return True
168 2
        elif value.lower() in self.false_values:
169 2
            return False
170 2
        raise ValueError(f"invalid value for boolean: {value!r}")
171

172 2
    def serialize(self, value, display=False):
173 2
        if value is True:
174 2
            return "true"
175 2
        elif value in (False, None):
176 2
            return "false"
177
        else:
178 2
            raise ValueError(f"{value!r} is not a boolean")
179

180

181 2
class List(ConfigValue):
182
    """List value.
183

184
    Supports elements split by commas or newlines. Newlines take presedence and
185
    empty list items will be filtered out.
186
    """
187

188 2
    def __init__(self, optional=False):
189 2
        self._required = not optional
190

191 2
    def deserialize(self, value):
192 2
        value = decode(value)
193 2
        if "\n" in value:
194 2
            values = re.split(r"\s*\n\s*", value)
195
        else:
196 2
            values = re.split(r"\s*,\s*", value)
197 2
        values = tuple(v.strip() for v in values if v.strip())
198 2
        validators.validate_required(values, self._required)
199 2
        return tuple(values)
200

201 2
    def serialize(self, value, display=False):
202 2
        if not value:
203 2
            return ""
204 2
        return "\n  " + "\n  ".join(encode(v) for v in value if v)
205

206

207 2
class LogColor(ConfigValue):
208 2
    def deserialize(self, value):
209 2
        value = decode(value)
210 2
        validators.validate_choice(value.lower(), log.COLORS)
211 2
        return value.lower()
212

213 2
    def serialize(self, value, display=False):
214 2
        if value.lower() in log.COLORS:
215 2
            return encode(value.lower())
216 2
        return ""
217

218

219 2
class LogLevel(ConfigValue):
220
    """Log level value.
221

222
    Expects one of ``critical``, ``error``, ``warning``, ``info``, ``debug``,
223
    or ``all``, with any casing.
224
    """
225

226 2
    levels = {
227
        "critical": logging.CRITICAL,
228
        "error": logging.ERROR,
229
        "warning": logging.WARNING,
230
        "info": logging.INFO,
231
        "debug": logging.DEBUG,
232
        "trace": log.TRACE_LOG_LEVEL,
233
        "all": logging.NOTSET,
234
    }
235

236 2
    def deserialize(self, value):
237 2
        value = decode(value)
238 2
        validators.validate_choice(value.lower(), self.levels.keys())
239 2
        return self.levels.get(value.lower())
240

241 2
    def serialize(self, value, display=False):
242 2
        lookup = {v: k for k, v in self.levels.items()}
243 2
        if value in lookup:
244 2
            return encode(lookup[value])
245 2
        return ""
246

247

248 2
class Hostname(ConfigValue):
249
    """Network hostname value."""
250

251 2
    def __init__(self, optional=False):
252 2
        self._required = not optional
253

254 2
    def deserialize(self, value, display=False):
255 2
        value = decode(value).strip()
256 2
        validators.validate_required(value, self._required)
257 2
        if not value:
258 2
            return None
259

260 2
        socket_path = path.get_unix_socket_path(value)
261 2
        if socket_path is not None:
262 2
            path_str = Path(not self._required).deserialize(socket_path)
263 2
            return f"unix:{path_str}"
264

265 2
        try:
266 2
            socket.getaddrinfo(value, None)
267 2
        except OSError:
268 2
            raise ValueError("must be a resolveable hostname or valid IP")
269

270 2
        return value
271

272

273 2
class Port(Integer):
274
    """Network port value.
275

276
    Expects integer in the range 0-65535, zero tells the kernel to simply
277
    allocate a port for us.
278
    """
279

280 2
    def __init__(self, choices=None, optional=False):
281 2
        super().__init__(
282
            minimum=0, maximum=2 ** 16 - 1, choices=choices, optional=optional
283
        )
284

285

286 2
class _ExpandedPath(str):
287 2
    def __new__(cls, original, expanded):
288 2
        return super().__new__(cls, expanded)
289

290 2
    def __init__(self, original, expanded):
291 2
        self.original = original
292

293

294 2
class Path(ConfigValue):
295
    """File system path.
296

297
    The following expansions of the path will be done:
298

299
    - ``~`` to the current user's home directory
300
    - ``$XDG_CACHE_DIR`` according to the XDG spec
301
    - ``$XDG_CONFIG_DIR`` according to the XDG spec
302
    - ``$XDG_DATA_DIR`` according to the XDG spec
303
    - ``$XDG_MUSIC_DIR`` according to the XDG spec
304
    """
305

306 2
    def __init__(self, optional=False):
307 2
        self._required = not optional
308

309 2
    def deserialize(self, value):
310 2
        value = decode(value).strip()
311 2
        expanded = path.expand_path(value)
312 2
        validators.validate_required(value, self._required)
313 2
        validators.validate_required(expanded, self._required)
314 2
        if not value or expanded is None:
315 2
            return None
316 2
        return _ExpandedPath(value, expanded)
317

318 2
    def serialize(self, value, display=False):
319 2
        if isinstance(value, _ExpandedPath):
320 2
            value = value.original
321 2
        if isinstance(value, bytes):
322 2
            value = value.decode(errors="surrogateescape")
323 2
        return value

Read our documentation on viewing source code .

Loading