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 decimal import Decimal
3import re
5from .exceptions import JsonSchemaValueException, JsonSchemaValuesException, JsonSchemaDefinitionException
6from .indent import indent
7from .ref_resolver import RefResolver
10def enforce_list(variable):
11 if isinstance(variable, list):
12 return variable
13 return [variable]
16# pylint: disable=too-many-instance-attributes,too-many-public-methods
17class CodeGenerator:
18 """
19 This class is not supposed to be used directly. Anything
20 inside of this class can be changed without noticing.
22 This class generates code of validation function from JSON
23 schema object as string. Example:
25 .. code-block:: python
27 CodeGenerator(json_schema_definition).func_code
28 """
30 INDENT = 4 # spaces
32 def __init__(self, definition, resolver=None, detailed_exceptions=True, fast_fail=True):
33 self._code = []
34 self._compile_regexps = {}
35 self._custom_formats = {}
36 self._detailed_exceptions = detailed_exceptions
37 self._fast_fail = fast_fail
39 # Any extra library should be here to be imported only once.
40 # Lines are imports to be printed in the file and objects
41 # key-value pair to pass to compile function directly.
42 self._extra_imports_lines = [
43 "from decimal import Decimal",
44 ]
45 self._extra_imports_objects = {
46 "Decimal": Decimal,
47 }
49 self._variables = {}
50 self._scope_stack = []
51 self._scope_counter = 0
52 self._last_closed_scope = None
53 self._indent = 0
54 self._indent_last_line = None
55 self._variable = None
56 self._variable_name = None
57 self._root_definition = definition
58 self._definition = None
60 # map schema URIs to validation function names for functions
61 # that are not yet generated, but need to be generated
62 self._needed_validation_functions = {}
63 # validation function names that are already done
64 self._validation_functions_done = set()
66 if resolver is None:
67 resolver = RefResolver.from_schema(definition, store={})
68 self._resolver = resolver
70 # add main function to `self._needed_validation_functions`
71 self._needed_validation_functions[self._resolver.get_uri()] = self._resolver.get_scope_name()
73 self._json_keywords_to_function = OrderedDict()
75 @property
76 def func_code(self):
77 """
78 Returns generated code of whole validation function as string.
79 """
80 self._generate_func_code()
82 return '\n'.join(self._code)
84 @property
85 def global_state(self):
86 """
87 Returns global variables for generating function from ``func_code``. Includes
88 compiled regular expressions and imports, so it does not have to do it every
89 time when validation function is called.
90 """
91 self._generate_func_code()
93 return dict(
94 **self._extra_imports_objects,
95 REGEX_PATTERNS=self._compile_regexps,
96 re=re,
97 JsonSchemaValueException=JsonSchemaValueException,
98 JsonSchemaValuesException=JsonSchemaValuesException,
99 )
101 @property
102 def global_state_code(self):
103 """
104 Returns global variables for generating function from ``func_code`` as code.
105 Includes compiled regular expressions and imports.
106 """
107 self._generate_func_code()
109 if not self._compile_regexps:
110 return '\n'.join(self._extra_imports_lines + [
111 'from fastjsonschema import JsonSchemaValueException, JsonSchemaValuesException',
112 '',
113 '',
114 ])
115 return '\n'.join(self._extra_imports_lines + [
116 'import re',
117 'from fastjsonschema import JsonSchemaValueException, JsonSchemaValuesException',
118 '',
119 '',
120 'REGEX_PATTERNS = ' + serialize_regexes(self._compile_regexps),
121 '',
122 ])
125 def _generate_func_code(self):
126 if not self._code:
127 self.generate_func_code()
129 def generate_func_code(self):
130 """
131 Creates base code of validation function and calls helper
132 for creating code by definition.
133 """
134 self.l('NoneType = type(None)')
135 # Generate parts that are referenced and not yet generated
136 while self._needed_validation_functions:
137 # During generation of validation function, could be needed to generate
138 # new one that is added again to `_needed_validation_functions`.
139 # Therefore usage of while instead of for loop.
140 uri, name = self._needed_validation_functions.popitem()
141 self.generate_validation_function(uri, name)
143 def generate_validation_function(self, uri, name):
144 """
145 Generate validation function for given uri with given name
146 """
147 self._validation_functions_done.add(uri)
148 self.l('')
149 with self._resolver.resolving(uri) as definition:
150 with self.l('def {}(data, custom_formats={{}}, name_prefix=None):', name):
151 if not self._fast_fail:
152 self.l('errors = []')
153 self.generate_func_code_block(definition, 'data', 'data', clear_variables=True)
154 if not self._fast_fail:
155 self.l('if errors: raise JsonSchemaValuesException(errors)')
156 self.l('return data')
158 def generate_func_code_block(self, definition, variable, variable_name, clear_variables=False):
159 """
160 Creates validation rules for current definition.
162 Returns the number of validation rules generated as code.
163 """
164 backup = self._definition, self._variable, self._variable_name
165 self._definition, self._variable, self._variable_name = definition, variable, variable_name
166 if clear_variables:
167 backup_variables = self._variables
168 self._variables = {}
170 count = self._generate_func_code_block(definition)
172 self._definition, self._variable, self._variable_name = backup
173 if clear_variables:
174 self._variables = backup_variables
176 return count
178 def _generate_func_code_block(self, definition):
179 if not isinstance(definition, dict):
180 raise JsonSchemaDefinitionException("definition must be an object")
181 if '$ref' in definition:
182 # needed because ref overrides any sibling keywords
183 return self.generate_ref()
184 return self.run_generate_functions(definition)
186 def run_generate_functions(self, definition):
187 """Returns the number of generate functions that were executed."""
188 count = 0
189 for key, func in self._json_keywords_to_function.items():
190 if key in definition:
191 func()
192 count += 1
193 return count
195 def generate_ref(self):
196 """
197 Ref can be link to remote or local definition.
199 .. code-block:: python
201 {'$ref': 'http://json-schema.org/draft-04/schema#'}
202 {
203 'properties': {
204 'foo': {'type': 'integer'},
205 'bar': {'$ref': '#/properties/foo'}
206 }
207 }
208 """
209 with self._resolver.in_scope(self._definition['$ref']):
210 name = self._resolver.get_scope_name()
211 uri = self._resolver.get_uri()
212 if uri not in self._validation_functions_done:
213 self._needed_validation_functions[uri] = name
214 # call validation function
215 assert self._variable_name.startswith("data")
216 path = self._variable_name[4:]
217 name_arg = '(name_prefix or "data") + "{}"'.format(path)
218 if '{' in name_arg:
219 name_arg = name_arg + '.format(**locals())'
220 self.l('{}({variable}, custom_formats, {name_arg})', name, name_arg=name_arg)
223 # pylint: disable=invalid-name
224 @indent
225 def l(self, line, *args, **kwds):
226 """
227 Short-cut of line. Used for inserting line. It's formated with parameters
228 ``variable``, ``variable_name`` (as ``name`` for short-cut), all keys from
229 current JSON schema ``definition`` and also passed arguments in ``args``
230 and named ``kwds``.
232 .. code-block:: python
234 self.l('if {variable} not in {enum}: raise JsonSchemaValueException("Wrong!")')
236 When you want to indent block, use it as context manager. For example:
238 .. code-block:: python
240 with self.l('if {variable} not in {enum}:'):
241 self.l('raise JsonSchemaValueException("Wrong!")')
242 """
243 spaces = ' ' * self.INDENT * self._indent
245 name = self._variable_name
246 if name:
247 # Add name_prefix to the name when it is being outputted.
248 assert name.startswith('data')
249 name = '" + (name_prefix or "data") + "' + name[4:]
250 if '{' in name:
251 name = name + '".format(**locals()) + "'
253 context = dict(
254 self._definition if self._definition and self._definition is not True else {},
255 variable=self._variable,
256 name=name,
257 **kwds
258 )
259 line = line.format(*args, **context)
260 line = line.replace('\n', '\\n').replace('\r', '\\r')
261 self._code.append(spaces + line)
262 return line
264 def e(self, string):
265 """
266 Short-cut of escape. Used for inserting user values into a string message.
268 .. code-block:: python
270 self.l('raise JsonSchemaValueException("Variable: {}")', self.e(variable))
271 """
272 if isinstance(string, str):
273 return string.encode('unicode_escape').decode('ascii').replace('"', '\\"')
274 return str(string).replace('"', '\\"')
276 def exc(self, msg, *args, append_to_msg=None, rule=None):
277 """
278 Short-cut for creating raising exception in the code.
279 """
280 if not self._detailed_exceptions:
281 if self._fast_fail:
282 self.l('raise JsonSchemaValueException("'+msg+'")', *args)
283 else:
284 self.l('errors.append(JsonSchemaValueException("'+msg+'"))', *args)
285 return
287 arg = '"'+msg+'"'
288 if append_to_msg:
289 arg += ' + (' + append_to_msg + ')'
290 # pylint: disable=line-too-long
291 msg = (
292 'raise JsonSchemaValueException('+arg+', value={variable}, name="{name}", definition={definition}, rule={rule})'
293 if self._fast_fail else
294 'errors.append(JsonSchemaValueException('+arg+', value={variable}, name="{name}", definition={definition}, rule={rule}))'
295 )
296 definition = self._expand_refs(self._definition)
297 definition_rule = self.e(definition.get(rule) if isinstance(definition, dict) else None)
298 self.l(msg, *args, definition=repr_default(definition), rule=repr(rule), definition_rule=definition_rule)
300 def _expand_refs(self, definition):
301 if isinstance(definition, list):
302 return [self._expand_refs(v) for v in definition]
303 if not isinstance(definition, dict):
304 return definition
305 if "$ref" in definition and isinstance(definition["$ref"], str):
306 with self._resolver.resolving(definition["$ref"]) as schema:
307 return schema
308 return {k: self._expand_refs(v) for k, v in definition.items()}
310 def _is_variable_in_scope(self, variable_name):
311 """
312 Whether ``variable_name`` was already defined in a block enclosing the
313 current one, and is therefore still bound here. A variable defined in a
314 sibling block is not, because that block may not have been entered.
315 """
316 scope = self._variables.get(variable_name)
317 if scope is None:
318 return False
319 return tuple(self._scope_stack[:len(scope)]) == scope
321 def create_variable_with_length(self):
322 """
323 Append code for creating variable with length of that variable
324 (for example length of list or dictionary) with name ``{variable}_len``.
325 It can be called several times and always it's done only when that variable
326 still does not exists.
327 """
328 variable_name = '{}_len'.format(self._variable)
329 if self._is_variable_in_scope(variable_name):
330 return
331 self._variables[variable_name] = tuple(self._scope_stack)
332 self.l('{variable}_len = len({variable})')
334 def create_variable_keys(self):
335 """
336 Append code for creating variable with keys of that variable (dictionary)
337 with a name ``{variable}_keys``. Similar to `create_variable_with_length`.
338 """
339 variable_name = '{}_keys'.format(self._variable)
340 if self._is_variable_in_scope(variable_name):
341 return
342 self._variables[variable_name] = tuple(self._scope_stack)
343 self.l('{variable}_keys = set({variable}.keys())')
345 def create_variable_is_list(self):
346 """
347 Append code for creating variable with bool if it's instance of list
348 with a name ``{variable}_is_list``. Similar to `create_variable_with_length`.
349 """
350 variable_name = '{}_is_list'.format(self._variable)
351 if self._is_variable_in_scope(variable_name):
352 return
353 self._variables[variable_name] = tuple(self._scope_stack)
354 self.l('{variable}_is_list = isinstance({variable}, (list, tuple))')
356 def create_variable_is_dict(self):
357 """
358 Append code for creating variable with bool if it's instance of list
359 with a name ``{variable}_is_dict``. Similar to `create_variable_with_length`.
360 """
361 variable_name = '{}_is_dict'.format(self._variable)
362 if self._is_variable_in_scope(variable_name):
363 return
364 self._variables[variable_name] = tuple(self._scope_stack)
365 self.l('{variable}_is_dict = isinstance({variable}, dict)')
368def serialize_regexes(patterns_dict):
369 # Unfortunately using `pprint.pformat` is causing errors
370 # specially with big regexes
371 regex_patterns = (
372 repr(k) + ": " + repr_regex(v)
373 for k, v in patterns_dict.items()
374 )
375 return '{\n ' + ",\n ".join(regex_patterns) + "\n}"
378def repr_default(value):
379 """
380 Like ``repr``, but renders non-finite floats as valid Python source.
382 ``repr(float('nan'))`` is ``'nan'``, which is not a name available in the
383 generated code, so a schema default of NaN or infinity has to be written
384 out as a ``float(...)`` call instead.
385 """
386 if isinstance(value, float) and (value != value or value in (float('inf'), float('-inf'))):
387 return "float({!r})".format(str(value))
388 if isinstance(value, list):
389 return '[' + ', '.join(repr_default(item) for item in value) + ']'
390 if isinstance(value, tuple):
391 return '(' + ''.join(repr_default(item) + ', ' for item in value) + ')'
392 if isinstance(value, dict):
393 return '{' + ', '.join(
394 '{}: {}'.format(repr_default(k), repr_default(v)) for k, v in value.items()
395 ) + '}'
396 return repr(value)
399def repr_regex(regex):
400 all_flags = ("A", "I", "DEBUG", "L", "M", "S", "X")
401 flags = " | ".join(f"re.{f}" for f in all_flags if regex.flags & getattr(re, f))
402 flags = ", " + flags if flags else ""
403 return "re.compile({!r}{})".format(regex.pattern, flags)