Pylons
Latest Version: 0.9.6.2
0001from interfaces import *
0002from api import *
0003import declarative
0004
0005__all__ = ['Schema']
0006
0007class Schema(FancyValidator):
0008
0009 """
0010 A schema validates a dictionary of values, applying different
0011 validators (be key) to the different values. If
0012 allow_extra_fields=True, keys without validators will be allowed;
0013 otherwise they will raise Invalid. If filter_extra_fields is
0014 set to true, then extra fields are not passed back in the results.
0015
0016 Validators are associated with keys either with a class syntax, or
0017 as keyword arguments (class syntax is usually easier). Something
0018 like::
0019
0020 class MySchema(Schema):
0021 name = Validators.PlainText()
0022 phone = Validators.PhoneNumber()
0023
0024 These will not be available as actual instance variables, but will
0025 be collected in a dictionary. To remove a validator in a subclass
0026 that is present in a superclass, set it to None, like::
0027
0028 class MySubSchema(MySchema):
0029 name = None
0030 """
0031
0032 # These validators will be applied before this schema:
0033 pre_validators = []
0034 # These validators will be applied after this schema:
0035 chained_validators = []
0036 # If true, then it is not an error when keys that aren't
0037 # associated with a validator are present:
0038 allow_extra_fields = False
0039 # If true, then keys that aren't associated with a validator
0040 # are removed:
0041 filter_extra_fields = False
0042 # If this is given, then any keys that aren't available but
0043 # are expected will be replaced with this value (and then
0044 # validated!) This does not override a present .if_missing
0045 # attribute on validators:
0046 if_key_missing = NoDefault
0047 # If true, then missing keys will be missing in the result,
0048 # if the validator doesn't have if_missing on it already:
0049 ignore_key_missing = False
0050 compound = True
0051 fields = {}
0052 order = []
0053
0054 messages = {
0055 'notExpected': 'The input field %(name)s was not expected.',
0056 'missingValue': "Missing value",
0057 }
0058
0059 __mutableattributes__ = ('fields', 'chained_validators',
0060 'pre_validators')
0061
0062 def __classinit__(cls, new_attrs):
0063 FancyValidator.__classinit__(cls, new_attrs)
0064 # Don't bother doing anything if this is the most parent
0065 # Schema class (which is the only class with just
0066 # FancyValidator as a superclass):
0067 if cls.__bases__ == (FancyValidator,):
0068 return cls
0069 # Scan through the class variables we've defined *just*
0070 # for this subclass, looking for validators (both classes
0071 # and instances):
0072 for key, value in new_attrs.items():
0073 if key in ('pre_validators', 'chained_validators',
0074 'view'):
0075 continue
0076 if is_validator(value):
0077 cls.fields[key] = value
0078 delattr(cls, key)
0079 # This last case means we're overwriting a validator
0080 # from a superclass:
0081 elif cls.fields.has_key(key):
0082 del cls.fields[key]
0083 for name, value in cls.fields.items():
0084 cls.add_field(name, value)
0085
0086 def __initargs__(self, new_attrs):
0087 for key, value in new_attrs.items():
0088 if key in ('pre_validators', 'chained_validators',
0089 'view'):
0090 continue
0091 if is_validator(value):
0092 self.fields[key] = value
0093 delattr(self, key)
0094 # This last case means we're overwriting a validator
0095 # from a superclass:
0096 elif self.fields.has_key(key):
0097 del self.fields[key]
0098 for name, value in self.fields.items():
0099 self.add_field(name, value)
0100
0101 def _to_python(self, value_dict, state):
0102 if not value_dict and self.if_empty is not NoDefault:
0103 return self.if_empty
0104
0105 for validator in self.pre_validators:
0106 value_dict = validator.to_python(value_dict, state)
0107
0108 new = {}
0109 errors = {}
0110 unused = self.fields.keys()
0111 if state is not None:
0112 previous_key = getattr(state, 'key', None)
0113 previous_full_dict = getattr(state, 'full_dict', None)
0114 state.full_dict = value_dict
0115 try:
0116 for name, value in value_dict.items():
0117 try:
0118 unused.remove(name)
0119 except ValueError:
0120 if not self.allow_extra_fields:
0121 raise Invalid(
0122 self.message('notExpected', state,
0123 name=repr(name)),
0124 value_dict, state)
0125 else:
0126 if not self.filter_extra_fields:
0127 new[name] = value
0128 continue
0129 validator = self.fields[name]
0130
0131 try:
0132 new[name] = validator.to_python(value, state)
0133 except Invalid, e:
0134 errors[name] = e
0135
0136 for name in unused:
0137 validator = self.fields[name]
0138 try:
0139 if_missing = validator.if_missing
0140 except AttributeError:
0141 if_missing = NoDefault
0142 if if_missing is NoDefault:
0143 if self.ignore_key_missing:
0144 continue
0145 if self.if_key_missing is NoDefault:
0146 errors[name] = Invalid(
0147 self.message('missingValue', state),
0148 None, state)
0149 else:
0150 try:
0151 new[name] = validator.to_python(self.if_key_missing, state)
0152 except Invalid, e:
0153 errors[name] = e
0154 else:
0155 new[name] = validator.if_missing
0156
0157 if errors:
0158 for validator in self.chained_validators:
0159 if (not hasattr(validator, 'validate_partial')
0160 or not getattr(validator, 'validate_partial_form', False)):
0161 continue
0162 try:
0163 validator.validate_partial(value_dict, state)
0164 except Invalid, e:
0165 sub_errors = e.unpack_errors()
0166 if not isinstance(sub_errors, dict):
0167 # Can't do anything here
0168 continue
0169 merge_dicts(errors, sub_errors)
0170
0171 if errors:
0172 raise Invalid(
0173 format_compound_error(errors),
0174 value_dict, state,
0175 error_dict=errors)
0176
0177 for validator in self.chained_validators:
0178 new = validator.to_python(new, state)
0179
0180 return new
0181
0182 finally:
0183 if state is not None:
0184 state.key = previous_key
0185 state.full_dict = previous_full_dict
0186
0187 def _from_python(self, value_dict, state):
0188 chained = self.chained_validators[:]
0189 chained.reverse()
0190 finished = []
0191 for validator in chained:
0192 __traceback_info__ = 'for_python chained_validator %s (finished %s)' % (validator, ', '.join(map(repr, finished)) or 'none')
0193 finished.append(validator)
0194 value_dict = validator.from_python(value_dict, state)
0195 new = {}
0196 errors = {}
0197 unused = self.fields.keys()
0198 if state is not None:
0199 previous_key = getattr(state, 'key', None)
0200 previous_full_dict = getattr(state, 'full_dict', None)
0201 state.full_dict = value_dict
0202 try:
0203 __traceback_info__ = None
0204 for name, value in value_dict.items():
0205 __traceback_info__ = 'for_python in %s' % name
0206 try:
0207 unused.remove(name)
0208 except ValueError:
0209 if not self.allow_extra_fields:
0210 raise Invalid(
0211 self.message('notExpected', state,
0212 name=repr(name)),
0213 value_dict, state)
0214 if not self.filter_extra_fields:
0215 new[name] = value
0216 else:
0217 try:
0218 new[name] = self.fields[name].from_python(value, state)
0219 except Invalid, e:
0220 errors[name] = e
0221
0222 del __traceback_info__
0223
0224 for name in unused:
0225 validator = self.fields[name]
0226 try:
0227 new[name] = validator.from_python(None, state)
0228 except Invalid, e:
0229 errors[name] = e
0230
0231 if errors:
0232 raise Invalid(
0233 format_compound_error(errors),
0234 value_dict, state,
0235 error_dict=errors)
0236
0237 pre = self.pre_validators[:]
0238 pre.reverse()
0239 for validator in pre:
0240 __traceback_info__ = 'for_python pre_validator %s' % validator
0241 new = validator.from_python(new, state)
0242
0243 return new
0244
0245 finally:
0246 if state is not None:
0247 state.key = previous_key
0248 state.full_dict = previous_full_dict
0249
0250
0251
0252 def add_chained_validator(self, cls, validator):
0253 if self is not None:
0254 if self.chained_validators is cls.chained_validators:
0255 self.chained_validators = cls.chained_validators[:]
0256 self.chained_validators.append(validator)
0257 else:
0258 cls.chained_validators.append(validator)
0259
0260 add_chained_validator = declarative.classinstancemethod(
0261 add_chained_validator)
0262
0263 def add_field(self, cls, name, validator):
0264 if self is not None:
0265 if self.fields is cls.fields:
0266 self.fields = cls.fields.copy()
0267 self.fields[name] = validator
0268 else:
0269 cls.fields[name] = validator
0270
0271 add_field = declarative.classinstancemethod(add_field)
0272
0273 def add_pre_validator(self, cls, validator):
0274 if self is not None:
0275 if self.pre_validators is cls.pre_validators:
0276 self.pre_validators = cls.pre_validators[:]
0277 self.pre_validators.append(validator)
0278 else:
0279 cls.pre_validators.append(validator)
0280
0281 add_pre_validator = declarative.classinstancemethod(add_pre_validator)
0282
0283 def subvalidators(self):
0284 result = []
0285 result.extend(self.pre_validators)
0286 result.extend(self.chained_validators)
0287 result.extend(self.fields.values())
0288 return result
0289
0290def format_compound_error(v, indent=0):
0291 if isinstance(v, Exception):
0292 try:
0293 return str(v)
0294 except (UnicodeDecodeError, UnicodeEncodeError):
0295 # There doesn't seem to be a better way to get a str()
0296 # version if possible, and unicode() if necessary, because
0297 # testing for the presence of a __unicode__ method isn't
0298 # enough
0299 return unicode(v)
0300 elif isinstance(v, dict):
0301 l = v.items()
0302 l.sort()
0303 return ('%s\n' % (' '*indent)).join(
0304 ["%s: %s" % (k, format_compound_error(value, indent=len(k)+2))
0305 for k, value in l
0306 if value is not None])
0307 elif isinstance(v, list):
0308 return ('%s\n' % (' '*indent)).join(
0309 ['%s' % (format_compound_error(value, indent=indent))
0310 for value in v
0311 if value is not None])
0312 elif isinstance(v, basestring):
0313 return v
0314 else:
0315 assert 0, "I didn't expect something like %s" % repr(v)
0316
0317def merge_dicts(d1, d2):
0318 for key in d2:
0319 if key in d1:
0320 d1[key] = merge_values(d1[key], d2[key])
0321 else:
0322 d1[key] = d2[key]
0323 return d1
0324
0325def merge_values(v1, v2):
0326 if (isinstance(v1, (str, unicode))
0327 and isinstance(v2, (str, unicode))):
0328 return v1 + '\n' + v2
0329 elif (isinstance(v1, (list, tuple))
0330 and isinstance(v2, (list, tuple))):
0331 return merge_lists(v1, v2)
0332 elif isinstance(v1, dict) and isinstance(v2, dict):
0333 return merge_dicts(v1, v2)
0334 else:
0335 # @@: Should we just ignore errors? Seems we do...
0336 return v1
0337
0338def merge_lists(l1, l2):
0339 if len(l1) < len(l2):
0340 l1 = l1 + [None]*(len(l2)-len(l1))
0341 elif len(l2) < len(l1):
0342 l2 = l2 + [None]*(len(l1)-len(l2))
0343 result = []
0344 for l1item, l2item in zip(l1, l2):
0345 item = None
0346 if l1item is None:
0347 item = l2item
0348 elif l2item is None:
0349 item = l1item
0350 else:
0351 item = merge_values(l1item, l2item)
0352 result.append(item)
0353 return result