Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from collections import OrderedDict
2from contextlib import contextmanager
3from decimal import Decimal
4import re
6from .exceptions import JsonSchemaValueException, JsonSchemaValuesException, JsonSchemaDefinitionException
7from .indent import indent
8from .ref_resolver import RefResolver
10# Both mean "this subschema did not match": a subschema behind a $ref becomes its own
11# function, which reports through JsonSchemaValuesException while errors are collected.
12VALIDATION_EXCEPTIONS = '(JsonSchemaValueException, JsonSchemaValuesException)'
15def enforce_list(variable):
16 if isinstance(variable, list):
17 return variable
18 return [variable]
21# pylint: disable=too-many-instance-attributes,too-many-public-methods
22class CodeGenerator:
23 """
24 This class is not supposed to be used directly. Anything
25 inside of this class can be changed without noticing.
27 This class generates code of validation function from JSON
28 schema object as string. Example:
30 .. code-block:: python
32 CodeGenerator(json_schema_definition).func_code
33 """
35 INDENT = 4 # spaces
37 def __init__(self, definition, resolver=None, detailed_exceptions=True, fast_fail=True):
38 self._code = []
39 self._compile_regexps = {}
40 self._custom_formats = {}
41 self._detailed_exceptions = detailed_exceptions
42 self._fast_fail = fast_fail
44 # Any extra library should be here to be imported only once.
45 # Lines are imports to be printed in the file and objects
46 # key-value pair to pass to compile function directly.
47 self._extra_imports_lines = [
48 "from decimal import Decimal",
49 ]
50 self._extra_imports_objects = {
51 "Decimal": Decimal,
52 }
54 self._variables = {}
55 self._scope_stack = []
56 self._scope_counter = 0
57 self._last_closed_scope = None
58 self._indent = 0
59 self._indent_last_line = None
60 self._variable = None
61 self._variable_name = None
62 self._root_definition = definition
63 self._definition = None
65 # map schema URIs to validation function names for functions
66 # that are not yet generated, but need to be generated
67 self._needed_validation_functions = {}
68 # validation function names that are already done
69 self._validation_functions_done = set()
71 if resolver is None:
72 resolver = RefResolver.from_schema(definition, store={})
73 self._resolver = resolver
75 # add main function to `self._needed_validation_functions`
76 self._needed_validation_functions[self._resolver.get_uri()] = self._resolver.get_scope_name()
78 self._json_keywords_to_function = OrderedDict()
80 @property
81 def func_code(self):
82 """
83 Returns generated code of whole validation function as string.
84 """
85 self._generate_func_code()
87 return '\n'.join(self._code)
89 @property
90 def global_state(self):
91 """
92 Returns global variables for generating function from ``func_code``. Includes
93 compiled regular expressions and imports, so it does not have to do it every
94 time when validation function is called.
95 """
96 self._generate_func_code()
98 return dict(
99 **self._extra_imports_objects,
100 REGEX_PATTERNS=self._compile_regexps,
101 re=re,
102 JsonSchemaValueException=JsonSchemaValueException,
103 JsonSchemaValuesException=JsonSchemaValuesException,
104 )
106 @property
107 def global_state_code(self):
108 """
109 Returns global variables for generating function from ``func_code`` as code.
110 Includes compiled regular expressions and imports.
111 """
112 self._generate_func_code()
114 if not self._compile_regexps:
115 return '\n'.join(self._extra_imports_lines + [
116 'from fastjsonschema import JsonSchemaValueException, JsonSchemaValuesException',
117 '',
118 '',
119 ])
120 return '\n'.join(self._extra_imports_lines + [
121 'import re',
122 'from fastjsonschema import JsonSchemaValueException, JsonSchemaValuesException',
123 '',
124 '',
125 'REGEX_PATTERNS = ' + serialize_regexes(self._compile_regexps),
126 '',
127 ])
130 def _generate_func_code(self):
131 if not self._code:
132 self.generate_func_code()
134 def generate_func_code(self):
135 """
136 Creates base code of validation function and calls helper
137 for creating code by definition.
138 """
139 self.l('NoneType = type(None)')
140 # Generate parts that are referenced and not yet generated
141 while self._needed_validation_functions:
142 # During generation of validation function, could be needed to generate
143 # new one that is added again to `_needed_validation_functions`.
144 # Therefore usage of while instead of for loop.
145 uri, name = self._needed_validation_functions.popitem()
146 self.generate_validation_function(uri, name)
148 def generate_validation_function(self, uri, name):
149 """
150 Generate validation function for given uri with given name
151 """
152 self._validation_functions_done.add(uri)
153 self.l('')
154 with self._resolver.resolving(uri) as definition:
155 with self.l('def {}(data, custom_formats={{}}, name_prefix=None):', name):
156 if not self._fast_fail:
157 self.l('errors = []')
158 self.generate_func_code_block(definition, 'data', 'data', clear_variables=True)
159 if not self._fast_fail:
160 self.l('if errors: raise JsonSchemaValuesException(errors)')
161 self.l('return data')
163 def generate_func_code_block(self, definition, variable, variable_name, clear_variables=False):
164 """
165 Creates validation rules for current definition.
167 Returns the number of validation rules generated as code.
168 """
169 backup = self._definition, self._variable, self._variable_name
170 self._definition, self._variable, self._variable_name = definition, variable, variable_name
171 if clear_variables:
172 backup_variables = self._variables
173 self._variables = {}
175 count = self._generate_func_code_block(definition)
177 self._definition, self._variable, self._variable_name = backup
178 if clear_variables:
179 self._variables = backup_variables
181 return count
183 @contextmanager
184 def trial_validation(self):
185 """
186 Subschemas of anyOf, oneOf, not, if, contains and propertyNames are only tried out.
187 Their failure is control flow for the surrounding ``try``, not an error to report,
188 so they have to raise even when ``fast_fail`` is off.
189 """
190 fast_fail, self._fast_fail = self._fast_fail, True
191 try:
192 yield
193 finally:
194 self._fast_fail = fast_fail
196 def _generate_func_code_block(self, definition):
197 if not isinstance(definition, dict):
198 raise JsonSchemaDefinitionException("definition must be an object")
199 if '$ref' in definition:
200 # needed because ref overrides any sibling keywords
201 return self.generate_ref()
202 return self.run_generate_functions(definition)
204 def run_generate_functions(self, definition):
205 """Returns the number of generate functions that were executed."""
206 count = 0
207 for key, func in self._json_keywords_to_function.items():
208 if key in definition:
209 func()
210 count += 1
211 return count
213 def generate_ref(self):
214 """
215 Ref can be link to remote or local definition.
217 .. code-block:: python
219 {'$ref': 'http://json-schema.org/draft-04/schema#'}
220 {
221 'properties': {
222 'foo': {'type': 'integer'},
223 'bar': {'$ref': '#/properties/foo'}
224 }
225 }
226 """
227 with self._resolver.in_scope(self._definition['$ref']):
228 name = self._resolver.get_scope_name()
229 uri = self._resolver.get_uri()
230 if uri not in self._validation_functions_done:
231 self._needed_validation_functions[uri] = name
232 # call validation function
233 assert self._variable_name.startswith("data")
234 path = self._variable_name[4:]
235 name_arg = '(name_prefix or "data") + "{}"'.format(path)
236 if '{' in name_arg:
237 name_arg = name_arg + '.format(**locals())'
238 if self._fast_fail:
239 self.l('{}({variable}, custom_formats, {name_arg})', name, name_arg=name_arg)
240 else:
241 # The referenced function collects into its own list, so merge it into ours
242 # instead of letting it abort the validation of the rest of the document.
243 with self.l('try:', optimize=False):
244 self.l('{}({variable}, custom_formats, {name_arg})', name, name_arg=name_arg)
245 with self.l('except JsonSchemaValuesException as e:'):
246 self.l('errors.extend(e.errors)')
249 # pylint: disable=invalid-name
250 @indent
251 def l(self, line, *args, **kwds):
252 """
253 Short-cut of line. Used for inserting line. It's formated with parameters
254 ``variable``, ``variable_name`` (as ``name`` for short-cut), all keys from
255 current JSON schema ``definition`` and also passed arguments in ``args``
256 and named ``kwds``.
258 .. code-block:: python
260 self.l('if {variable} not in {enum}: raise JsonSchemaValueException("Wrong!")')
262 When you want to indent block, use it as context manager. For example:
264 .. code-block:: python
266 with self.l('if {variable} not in {enum}:'):
267 self.l('raise JsonSchemaValueException("Wrong!")')
268 """
269 spaces = ' ' * self.INDENT * self._indent
271 name = self._variable_name
272 if name:
273 # Add name_prefix to the name when it is being outputted.
274 assert name.startswith('data')
275 name = '" + (name_prefix or "data") + "' + name[4:]
276 if '{' in name:
277 name = name + '".format(**locals()) + "'
279 context = dict(
280 self._definition if self._definition and self._definition is not True else {},
281 variable=self._variable,
282 name=name,
283 **kwds
284 )
285 line = line.format(*args, **context)
286 line = line.replace('\n', '\\n').replace('\r', '\\r')
287 self._code.append(spaces + line)
288 return line
290 def e(self, string):
291 """
292 Short-cut of escape. Used for inserting user values into a string message.
294 .. code-block:: python
296 self.l('raise JsonSchemaValueException("Variable: {}")', self.e(variable))
297 """
298 if isinstance(string, str):
299 return string.encode('unicode_escape').decode('ascii').replace('"', '\\"')
300 return str(string).replace('"', '\\"')
302 def exc(self, msg, *args, append_to_msg=None, rule=None):
303 """
304 Short-cut for creating raising exception in the code.
305 """
306 if not self._detailed_exceptions:
307 if self._fast_fail:
308 self.l('raise JsonSchemaValueException("'+msg+'")', *args)
309 else:
310 self.l('errors.append(JsonSchemaValueException("'+msg+'"))', *args)
311 return
313 arg = '"'+msg+'"'
314 if append_to_msg:
315 arg += ' + (' + append_to_msg + ')'
316 # pylint: disable=line-too-long
317 msg = (
318 'raise JsonSchemaValueException('+arg+', value={variable}, name="{name}", definition={definition}, rule={rule})'
319 if self._fast_fail else
320 'errors.append(JsonSchemaValueException('+arg+', value={variable}, name="{name}", definition={definition}, rule={rule}))'
321 )
322 definition = self._expand_refs(self._definition)
323 definition_rule = self.e(definition.get(rule) if isinstance(definition, dict) else None)
324 self.l(msg, *args, definition=repr_default(definition), rule=repr(rule), definition_rule=definition_rule)
326 def _expand_refs(self, definition):
327 if isinstance(definition, list):
328 return [self._expand_refs(v) for v in definition]
329 if not isinstance(definition, dict):
330 return definition
331 if "$ref" in definition and isinstance(definition["$ref"], str):
332 with self._resolver.resolving(definition["$ref"]) as schema:
333 return schema
334 return {k: self._expand_refs(v) for k, v in definition.items()}
336 def _is_variable_in_scope(self, variable_name):
337 """
338 Whether ``variable_name`` was already defined in a block enclosing the
339 current one, and is therefore still bound here. A variable defined in a
340 sibling block is not, because that block may not have been entered.
341 """
342 scope = self._variables.get(variable_name)
343 if scope is None:
344 return False
345 return tuple(self._scope_stack[:len(scope)]) == scope
347 def create_variable_with_length(self):
348 """
349 Append code for creating variable with length of that variable
350 (for example length of list or dictionary) with name ``{variable}_len``.
351 It can be called several times and always it's done only when that variable
352 still does not exists.
353 """
354 variable_name = '{}_len'.format(self._variable)
355 if self._is_variable_in_scope(variable_name):
356 return
357 self._variables[variable_name] = tuple(self._scope_stack)
358 self.l('{variable}_len = len({variable})')
360 def create_variable_keys(self):
361 """
362 Append code for creating variable with keys of that variable (dictionary)
363 with a name ``{variable}_keys``. Similar to `create_variable_with_length`.
364 """
365 variable_name = '{}_keys'.format(self._variable)
366 if self._is_variable_in_scope(variable_name):
367 return
368 self._variables[variable_name] = tuple(self._scope_stack)
369 self.l('{variable}_keys = set({variable}.keys())')
371 def create_variable_is_list(self):
372 """
373 Append code for creating variable with bool if it's instance of list
374 with a name ``{variable}_is_list``. Similar to `create_variable_with_length`.
375 """
376 variable_name = '{}_is_list'.format(self._variable)
377 if self._is_variable_in_scope(variable_name):
378 return
379 self._variables[variable_name] = tuple(self._scope_stack)
380 self.l('{variable}_is_list = isinstance({variable}, (list, tuple))')
382 def create_variable_is_dict(self):
383 """
384 Append code for creating variable with bool if it's instance of list
385 with a name ``{variable}_is_dict``. Similar to `create_variable_with_length`.
386 """
387 variable_name = '{}_is_dict'.format(self._variable)
388 if self._is_variable_in_scope(variable_name):
389 return
390 self._variables[variable_name] = tuple(self._scope_stack)
391 self.l('{variable}_is_dict = isinstance({variable}, dict)')
394def serialize_regexes(patterns_dict):
395 # Unfortunately using `pprint.pformat` is causing errors
396 # specially with big regexes
397 regex_patterns = (
398 repr(k) + ": " + repr_regex(v)
399 for k, v in patterns_dict.items()
400 )
401 return '{\n ' + ",\n ".join(regex_patterns) + "\n}"
404def repr_default(value):
405 """
406 Like ``repr``, but renders non-finite floats as valid Python source.
408 ``repr(float('nan'))`` is ``'nan'``, which is not a name available in the
409 generated code, so a schema default of NaN or infinity has to be written
410 out as a ``float(...)`` call instead.
411 """
412 if isinstance(value, float) and (value != value or value in (float('inf'), float('-inf'))):
413 return "float({!r})".format(str(value))
414 if isinstance(value, list):
415 return '[' + ', '.join(repr_default(item) for item in value) + ']'
416 if isinstance(value, tuple):
417 return '(' + ''.join(repr_default(item) + ', ' for item in value) + ')'
418 if isinstance(value, dict):
419 return '{' + ', '.join(
420 '{}: {}'.format(repr_default(k), repr_default(v)) for k, v in value.items()
421 ) + '}'
422 return repr(value)
425def repr_regex(regex):
426 all_flags = ("A", "I", "DEBUG", "L", "M", "S", "X")
427 flags = " | ".join(f"re.{f}" for f in all_flags if regex.flags & getattr(re, f))
428 flags = ", " + flags if flags else ""
429 return "re.compile({!r}{})".format(regex.pattern, flags)