1 2
from itertools import chain
2 2
from types import MappingProxyType
3

4 2
from dynaconf import validator_conditions  # noqa
5 2
from dynaconf.utils.functional import empty
6

7

8 2
EQUALITY_ATTRS = (
9
    "names",
10
    "must_exist",
11
    "when",
12
    "condition",
13
    "operations",
14
    "envs",
15
)
16

17

18 2
class ValidationError(Exception):
19 2
    pass
20

21

22 2
class Validator(object):
23
    """Validators are conditions attached to settings variables names
24
    or patterns::
25

26
        Validator('MESSAGE', must_exist=True, eq='Hello World')
27

28
    The above ensure MESSAGE is available in default env and
29
    is equal to 'Hello World'
30

31
    `names` are a one (or more) names or patterns::
32

33
        Validator('NAME')
34
        Validator('NAME', 'OTHER_NAME', 'EVEN_OTHER')
35
        Validator(r'^NAME', r'OTHER./*')
36

37
    The `operations` are::
38

39
        eq: value == other
40
        ne: value != other
41
        gt: value > other
42
        lt: value < other
43
        gte: value >= other
44
        lte: value <= other
45
        is_type_of: isinstance(value, type)
46
        is_in:  value in sequence
47
        is_not_in: value not in sequence
48
        identity: value is other
49
        cont: contain value in
50
        len_eq: len(value) == other
51
        len_ne: len(value) != other
52
        len_min: len(value) > other
53
        len_max: len(value) < other
54

55
    `env` is which env to be checked, can be a list or
56
    default is used.
57

58
    `when` holds a validator and its return decides if validator runs or not::
59

60
        Validator('NAME', must_exist=True, when=Validator('OTHER', eq=2))
61
        # NAME is required only if OTHER eq to 2
62
        # When the very first thing to be performed when passed.
63
        # if no env is passed to `when` it is inherited
64

65
    `must_exist` is alias to `required` requirement. (executed after when)::
66

67
       settings.get(value, empty) returns non empty
68

69
    condition is a callable to be executed and return boolean::
70

71
       Validator('NAME', condition=lambda x: x == 1)
72
       # it is executed before operations.
73

74
    """
75

76 2
    default_messages = MappingProxyType(
77
        {
78
            "must_exist_true": "{name} is required in env {env}",
79
            "must_exist_false": "{name} cannot exists in env {env}",
80
            "condition": "{name} invalid for {function}({value}) in env {env}",
81
            "operations": (
82
                "{name} must {operation} {op_value} "
83
                "but it is {value} in env {env}"
84
            ),
85
            "combined": "combined validators failed {errors}",
86
        }
87
    )
88

89 2
    def __init__(
90
        self,
91
        *names,
92
        must_exist=None,
93
        required=None,  # this is alias for `must_exist`
94
        condition=None,
95
        when=None,
96
        env=None,
97
        messages=None,
98
        cast=None,
99
        default=empty,  # Literal value or a callable
100
        **operations
101
    ):
102
        # Copy immutable MappingProxyType as a mutable dict
103 2
        self.messages = dict(self.default_messages)
104 2
        if messages:
105 2
            self.messages.update(messages)
106

107 2
        if when is not None and not isinstance(when, Validator):
108 2
            raise TypeError("when must be Validator instance")
109

110 2
        if condition is not None and not callable(condition):
111 2
            raise TypeError("condition must be callable")
112

113 2
        self.names = names
114 2
        self.must_exist = must_exist if must_exist is not None else required
115 2
        self.condition = condition
116 2
        self.when = when
117 2
        self.cast = cast or (lambda value: value)
118 2
        self.operations = operations
119 2
        self.default = default
120

121 2
        if isinstance(env, str):
122 2
            self.envs = [env]
123 2
        elif isinstance(env, (list, tuple)):
124 2
            self.envs = env
125
        else:
126 2
            self.envs = None
127

128 2
    def __or__(self, other):
129 2
        return OrValidator(self, other)
130

131 2
    def __and__(self, other):
132 2
        return AndValidator(self, other)
133

134 2
    def __eq__(self, other):
135 2
        if self is other:
136 2
            return True
137

138 2
        if type(self).__name__ != type(other).__name__:
139 2
            return False
140

141 2
        identical_attrs = (
142
            getattr(self, attr) == getattr(other, attr)
143
            for attr in EQUALITY_ATTRS
144
        )
145 2
        if all(identical_attrs):
146 2
            return True
147

148 2
        return False
149

150 2
    def validate(self, settings):
151
        """Raise ValidationError if invalid"""
152

153 2
        if self.envs is None:
154 2
            self.envs = [settings.current_env]
155

156 2
        if self.when is not None:
157 2
            try:
158
                # inherit env if not defined
159 2
                if self.when.envs is None:
160 2
                    self.when.envs = self.envs
161

162 2
                self.when.validate(settings)
163 2
            except ValidationError:
164
                # if when is invalid, return canceling validation flow
165 2
                return
166

167
        # If only using current_env, skip using_env decoration (reload)
168 2
        if (
169
            len(self.envs) == 1
170
            and self.envs[0].upper() == settings.current_env.upper()
171
        ):
172 2
            self._validate_items(settings, settings.current_env)
173 2
            return
174

175 2
        for env in self.envs:
176 2
            self._validate_items(settings.from_env(env))
177

178 2
    def _validate_items(self, settings, env=None):
179 2
        env = env or settings.current_env
180 2
        for name in self.names:
181 2
            if self.default is not empty:
182 2
                default_value = (
183
                    self.default(settings, self)
184
                    if callable(self.default)
185
                    else self.default
186
                )
187
            else:
188 2
                default_value = empty
189

190 2
            value = self.cast(settings.setdefault(name, default_value))
191

192
            # is name required but not exists?
193 2
            if self.must_exist is True and value is empty:
194 2
                raise ValidationError(
195
                    self.messages["must_exist_true"].format(name=name, env=env)
196
                )
197 2
            elif self.must_exist is False and value is not empty:
198 2
                raise ValidationError(
199
                    self.messages["must_exist_false"].format(
200
                        name=name, env=env
201
                    )
202
                )
203 2
            elif self.must_exist in (False, None) and value is empty:
204 2
                continue
205

206
            # is there a callable condition?
207 2
            if self.condition is not None:
208 2
                if not self.condition(value):
209 2
                    raise ValidationError(
210
                        self.messages["condition"].format(
211
                            name=name,
212
                            function=self.condition.__name__,
213
                            value=value,
214
                            env=env,
215
                        )
216
                    )
217

218
            # operations
219 2
            for op_name, op_value in self.operations.items():
220 2
                op_function = getattr(validator_conditions, op_name)
221 2
                if not op_function(value, op_value):
222 2
                    raise ValidationError(
223
                        self.messages["operations"].format(
224
                            name=name,
225
                            operation=op_function.__name__,
226
                            op_value=op_value,
227
                            value=value,
228
                            env=env,
229
                        )
230
                    )
231

232

233 2
class CombinedValidator(Validator):
234 2
    def __init__(self, validator_a, validator_b, *args, **kwargs):
235
        """Takes 2 validators and combines the validation"""
236 2
        self.validators = (validator_a, validator_b)
237 2
        super().__init__(*args, **kwargs)
238 2
        for attr in EQUALITY_ATTRS:
239 2
            if not getattr(self, attr, None):
240 2
                value = tuple(
241
                    getattr(validator, attr) for validator in self.validators
242
                )
243 2
                setattr(self, attr, value)
244

245
    def validate(self, settings):  # pragma: no cover
246
        raise NotImplementedError(
247
            "subclasses OrValidator or AndValidator implements this method"
248
        )
249

250

251 2
class OrValidator(CombinedValidator):
252
    """Evaluates on Validator() | Validator()"""
253

254 2
    def validate(self, settings):
255
        """Ensure at least one of the validators are valid"""
256 2
        errors = []
257 2
        for validator in self.validators:
258 2
            try:
259 2
                validator.validate(settings)
260 2
            except ValidationError as e:
261 2
                errors.append(e)
262 2
                continue
263
            else:
264 2
                return
265

266 2
        raise ValidationError(
267
            self.messages["combined"].format(
268
                errors=" or ".join(
269
                    str(e).replace("combined validators failed ", "")
270
                    for e in errors
271
                )
272
            )
273
        )
274

275

276 2
class AndValidator(CombinedValidator):
277
    """Evaluates on Validator() & Validator()"""
278

279 2
    def validate(self, settings):
280
        """Ensure both the validators are valid"""
281 2
        errors = []
282 2
        for validator in self.validators:
283 2
            try:
284 2
                validator.validate(settings)
285 2
            except ValidationError as e:
286 2
                errors.append(e)
287 2
                continue
288

289 2
        if errors:
290 2
            raise ValidationError(
291
                self.messages["combined"].format(
292
                    errors=" and ".join(
293
                        str(e).replace("combined validators failed ", "")
294
                        for e in errors
295
                    )
296
                )
297
            )
298

299

300 2
class ValidatorList(list):
301 2
    def __init__(self, settings, validators=None, *args, **kwargs):
302 2
        if isinstance(validators, (list, tuple)):
303 2
            args = list(args) + list(validators)
304 2
        super(ValidatorList, self).__init__(args, **kwargs)
305 2
        self.settings = settings
306

307 2
    def register(self, *args, **kwargs):
308 2
        validators = list(chain.from_iterable(kwargs.values()))
309 2
        validators.extend(args)
310 2
        for validator in validators:
311 2
            if validator and validator not in self:
312 2
                self.append(validator)
313

314 2
    def validate(self):
315 2
        for validator in self:
316 2
            validator.validate(self.settings)

Read our documentation on viewing source code .

Loading