st2common.util.schema.is_property_type_list() - Code Metrics - Inspection of "Fix explicit none to optional argument on action e..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#2335)

by W

created 2016-01-06 03:25 UTC

st2common.util.schema.is_property_type_list() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	3
rs	10

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import copy

import six
import jsonschema
from jsonschema import _validators
from jsonschema.validators import create

from st2common.util import jsonify

__all__ = [
    'get_validator',
    'get_draft_schema',
    'get_action_parameters_schema',
    'get_schema_for_action_parameters',
    'get_schema_for_resource_parameters',
    'is_property_type_single',
    'is_property_type_list',
    'is_property_type_anyof',
    'is_property_type_oneof',
    'is_property_nullable',
    'is_attribute_type_array',
    'is_attribute_type_object',
    'validate'
]

# https://github.com/json-schema/json-schema/blob/master/draft-04/schema
# The source material is licensed under the AFL or BSD license.
# Both draft 4 and custom schema has additionalProperties set to false by default.
# The custom schema differs from draft 4 with the extension of position, immutable,
# and draft 3 version of required.
PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)))
SCHEMAS = {
    'draft4': jsonify.load_file(os.path.join(PATH, 'draft4.json')),
    'custom': jsonify.load_file(os.path.join(PATH, 'custom.json')),

    # Custom schema for action params which doesn't allow parameter "type" attribute to be array
    'action_params': jsonify.load_file(os.path.join(PATH, 'action_params.json'))
}

SCHEMA_ANY_TYPE = {
    "anyOf": [
        {"type": "array"},
        {"type": "boolean"},
        {"type": "integer"},
        {"type": "number"},
        {"type": "object"},
        {"type": "string"}
    ]
}


def get_draft_schema(version='custom', additional_properties=False):
    schema = copy.deepcopy(SCHEMAS[version])
    if additional_properties and 'additionalProperties' in schema:
        del schema['additionalProperties']
    return schema


def get_action_parameters_schema(additional_properties=False):
    """
    Return a generic schema which is used for validating action parameters definition.
    """
    return get_draft_schema(version='action_params', additional_properties=additional_properties)


CustomValidator = create(
    meta_schema=get_draft_schema(version='custom', additional_properties=True),
    validators={
        u"$ref": _validators.ref,
        u"additionalItems": _validators.additionalItems,
        u"additionalProperties": _validators.additionalProperties,
        u"allOf": _validators.allOf_draft4,
        u"anyOf": _validators.anyOf_draft4,
        u"dependencies": _validators.dependencies,
        u"enum": _validators.enum,
        u"format": _validators.format,
        u"items": _validators.items,
        u"maxItems": _validators.maxItems,
        u"maxLength": _validators.maxLength,
        u"maxProperties": _validators.maxProperties_draft4,
        u"maximum": _validators.maximum,
        u"minItems": _validators.minItems,
        u"minLength": _validators.minLength,
        u"minProperties": _validators.minProperties_draft4,
        u"minimum": _validators.minimum,
        u"multipleOf": _validators.multipleOf,
        u"not": _validators.not_draft4,
        u"oneOf": _validators.oneOf_draft4,
        u"pattern": _validators.pattern,
        u"patternProperties": _validators.patternProperties,
        u"properties": _validators.properties_draft3,
        u"type": _validators.type_draft4,
        u"uniqueItems": _validators.uniqueItems,
    },
    version="custom_validator",
)


def is_property_type_single(property_schema):
    return (isinstance(property_schema, dict) and
            'anyOf' not in property_schema.keys() and
            'oneOf' not in property_schema.keys() and
            not isinstance(property_schema.get('type', 'string'), list))


def is_property_type_list(property_schema):
    return (isinstance(property_schema, dict) and
            isinstance(property_schema.get('type', 'string'), list))


def is_property_type_anyof(property_schema):
    return isinstance(property_schema, dict) and 'anyOf' in property_schema.keys()


def is_property_type_oneof(property_schema):
    return isinstance(property_schema, dict) and 'oneOf' in property_schema.keys()


def is_property_nullable(property_type_schema):
    # For anyOf and oneOf, the property_schema is a list of types.
    if isinstance(property_type_schema, list):
        return len([t for t in property_type_schema
                    if ((isinstance(t, six.string_types) and t == 'null') or
                        (isinstance(t, dict) and t.get('type', 'string') == 'null'))]) > 0

    return (isinstance(property_type_schema, dict) and
            property_type_schema.get('type', 'string') == 'null')


def is_attribute_type_array(attribute_type):
    return (attribute_type == 'array' or
            (isinstance(attribute_type, list) and 'array' in attribute_type))


def is_attribute_type_object(attribute_type):
    return (attribute_type == 'object' or
            (isinstance(attribute_type, list) and 'object' in attribute_type))


def assign_default_values(instance, schema):
    """
    Assign default values on the provided instance based on the schema default specification.
    """
    instance = copy.deepcopy(instance)
    instance_is_dict = isinstance(instance, dict)
    instance_is_array = isinstance(instance, list)

    if not instance_is_dict and not instance_is_array:
        return instance

    properties = schema.get('properties', {})

    for property_name, property_data in six.iteritems(properties):
        has_default_value = 'default' in property_data
        default_value = property_data.get('default', None)

        # Assign default value on the instance so the validation doesn't fail if requires is true
        # but the value is not provided
        if has_default_value:
            if instance_is_dict and instance.get(property_name, None) is None:
                instance[property_name] = default_value
            elif instance_is_array:
                for index, _ in enumerate(instance):
                    if instance[index].get(property_name, None) is None:
                        instance[index][property_name] = default_value

        # Support for nested properties (array and object)
        attribute_type = property_data.get('type', None)
        schema_items = property_data.get('items', {})

        # Array
        if (is_attribute_type_array(attribute_type) and
                schema_items and schema_items.get('properties', {})):
            array_instance = instance.get(property_name, None)
            array_schema = schema['properties'][property_name]['items']

            if array_instance is not None:
                # Note: We don't perform subschema assignment if no value is provided
                instance[property_name] = assign_default_values(instance=array_instance,
                                                                schema=array_schema)

        # Object
        if is_attribute_type_object(attribute_type) and property_data.get('properties', {}):
            object_instance = instance.get(property_name, None)
            object_schema = schema['properties'][property_name]

            if object_instance is not None:
                # Note: We don't perform subschema assignment if no value is provided
                instance[property_name] = assign_default_values(instance=object_instance,
                                                                schema=object_schema)

    return instance


def modify_schema_allow_default_none(schema):
    """
    Manipulate the provided schema so None is also an allowed value for each attribute which
    defines a default value of None.
    """
    schema = copy.deepcopy(schema)
    properties = schema.get('properties', {})

    for property_name, property_data in six.iteritems(properties):
        is_optional = not property_data.get('required', False)
        has_default_value = 'default' in property_data
        default_value = property_data.get('default', None)
        property_schema = schema['properties'][property_name]

        if (has_default_value or is_optional) and default_value is None:
            # If property is anyOf and oneOf then it has to be process differently.
            if (is_property_type_anyof(property_schema) and
                    not is_property_nullable(property_schema['anyOf'])):
                property_schema['anyOf'].append({'type': 'null'})
            elif (is_property_type_oneof(property_schema) and
                    not is_property_nullable(property_schema['oneOf'])):
                property_schema['oneOf'].append({'type': 'null'})
            elif (is_property_type_list(property_schema) and
                    not is_property_nullable(property_schema.get('type'))):
                property_schema['type'].append('null')
            elif (is_property_type_single(property_schema) and
                    not is_property_nullable(property_schema.get('type'))):
                property_schema['type'] = [property_schema.get('type', 'string'), 'null']

        # Support for nested properties (array and object)
        attribute_type = property_data.get('type', None)
        schema_items = property_data.get('items', {})

        # Array
        if (is_attribute_type_array(attribute_type) and
                schema_items and schema_items.get('properties', {})):
            array_schema = schema_items
            array_schema = modify_schema_allow_default_none(schema=array_schema)
            schema['properties'][property_name]['items'] = array_schema

        # Object
        if is_attribute_type_object(attribute_type) and property_data.get('properties', {}):
            object_schema = property_data
            object_schema = modify_schema_allow_default_none(schema=object_schema)
            schema['properties'][property_name] = object_schema

    return schema


def validate(instance, schema, cls=None, use_default=True, allow_default_none=False, *args,
             **kwargs):
    """
    Custom validate function which supports default arguments combined with the "required"
    property.

    Note: This function returns cleaned instance with default values assigned.

    :param use_default: True to support the use of the optional "default" property.
    :type use_default: ``bool``
    """
    instance = copy.deepcopy(instance)
    schema_type = schema.get('type', None)
    instance_is_dict = isinstance(instance, dict)

    if use_default and allow_default_none:
        schema = modify_schema_allow_default_none(schema=schema)

    if use_default and schema_type == 'object' and instance_is_dict:
        instance = assign_default_values(instance=instance, schema=schema)

    # pylint: disable=assignment-from-no-return
    jsonschema.validate(instance=instance, schema=schema, cls=cls, *args, **kwargs)

    return instance


VALIDATORS = {
    'draft4': jsonschema.Draft4Validator,
    'custom': CustomValidator
}


def get_validator(version='custom'):
    validator = VALIDATORS[version]
    return validator


def get_schema_for_action_parameters(action_db):
    """
    Dynamically construct JSON schema for the provided action from the parameters metadata.

    Note: This schema is used to validate parameters which are passed to the action.
    """
    from st2common.util.action_db import get_runnertype_by_name
    runner_type = get_runnertype_by_name(action_db.runner_type['name'])

    parameters_schema = {}
    parameters_schema.update(runner_type.runner_parameters)
    parameters_schema.update(action_db.parameters)

    schema = get_schema_for_resource_parameters(parameters_schema=parameters_schema)

    if parameters_schema:
        schema['title'] = action_db.name
        if action_db.description:
            schema['description'] = action_db.description

    return schema


def get_schema_for_resource_parameters(parameters_schema):
    """
    Dynamically construct JSON schema for the provided resource from the parameters metadata.
    """
    def normalize(x):
        return {k: v if v else SCHEMA_ANY_TYPE for k, v in six.iteritems(x)}

    schema = {}
    properties = {}
    properties.update(normalize(parameters_schema))
    if properties:
        schema['type'] = 'object'
        schema['properties'] = properties
        schema['additionalProperties'] = False

    return schema


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
17			import copy
18
19			import six
20			import jsonschema
21			from jsonschema import _validators
22			from jsonschema.validators import create
23
24			from st2common.util import jsonify
25
26			__all__ = [
27			'get_validator',
28			'get_draft_schema',
29			'get_action_parameters_schema',
30			'get_schema_for_action_parameters',
31			'get_schema_for_resource_parameters',
32			'is_property_type_single',
33			'is_property_type_list',
34			'is_property_type_anyof',
35			'is_property_type_oneof',
36			'is_property_nullable',
37			'is_attribute_type_array',
38			'is_attribute_type_object',
39			'validate'
40			]
41
42			# https://github.com/json-schema/json-schema/blob/master/draft-04/schema
43			# The source material is licensed under the AFL or BSD license.
44			# Both draft 4 and custom schema has additionalProperties set to false by default.
45			# The custom schema differs from draft 4 with the extension of position, immutable,
46			# and draft 3 version of required.
47			PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)))
48			SCHEMAS = {
49			'draft4': jsonify.load_file(os.path.join(PATH, 'draft4.json')),
50			'custom': jsonify.load_file(os.path.join(PATH, 'custom.json')),
51
52			# Custom schema for action params which doesn't allow parameter "type" attribute to be array
53			'action_params': jsonify.load_file(os.path.join(PATH, 'action_params.json'))
54			}
55
56			SCHEMA_ANY_TYPE = {
57			"anyOf": [
58			{"type": "array"},
59			{"type": "boolean"},
60			{"type": "integer"},
61			{"type": "number"},
62			{"type": "object"},
63			{"type": "string"}
64			]
65			}
66
67
68			def get_draft_schema(version='custom', additional_properties=False):
69			schema = copy.deepcopy(SCHEMAS[version])
70			if additional_properties and 'additionalProperties' in schema:
71			del schema['additionalProperties']
72			return schema
73
74
75			def get_action_parameters_schema(additional_properties=False):
76			"""
77			Return a generic schema which is used for validating action parameters definition.
78			"""
79			return get_draft_schema(version='action_params', additional_properties=additional_properties)
80
81
82			CustomValidator = create(
83			meta_schema=get_draft_schema(version='custom', additional_properties=True),
84			validators={
85			u"$ref": _validators.ref,
86			u"additionalItems": _validators.additionalItems,
87			u"additionalProperties": _validators.additionalProperties,
88			u"allOf": _validators.allOf_draft4,
89			u"anyOf": _validators.anyOf_draft4,
90			u"dependencies": _validators.dependencies,
91			u"enum": _validators.enum,
92			u"format": _validators.format,
93			u"items": _validators.items,
94			u"maxItems": _validators.maxItems,
95			u"maxLength": _validators.maxLength,
96			u"maxProperties": _validators.maxProperties_draft4,
97			u"maximum": _validators.maximum,
98			u"minItems": _validators.minItems,
99			u"minLength": _validators.minLength,
100			u"minProperties": _validators.minProperties_draft4,
101			u"minimum": _validators.minimum,
102			u"multipleOf": _validators.multipleOf,
103			u"not": _validators.not_draft4,
104			u"oneOf": _validators.oneOf_draft4,
105			u"pattern": _validators.pattern,
106			u"patternProperties": _validators.patternProperties,
107			u"properties": _validators.properties_draft3,
108			u"type": _validators.type_draft4,
109			u"uniqueItems": _validators.uniqueItems,
110			},
111			version="custom_validator",
112			)
113
114
115			def is_property_type_single(property_schema):
116			return (isinstance(property_schema, dict) and
117			'anyOf' not in property_schema.keys() and
118			'oneOf' not in property_schema.keys() and
119			not isinstance(property_schema.get('type', 'string'), list))
120
121
122			def is_property_type_list(property_schema):
123			return (isinstance(property_schema, dict) and
124			isinstance(property_schema.get('type', 'string'), list))
125
126
127			def is_property_type_anyof(property_schema):
128			return isinstance(property_schema, dict) and 'anyOf' in property_schema.keys()
129
130
131			def is_property_type_oneof(property_schema):
132			return isinstance(property_schema, dict) and 'oneOf' in property_schema.keys()
133
134
135			def is_property_nullable(property_type_schema):
136			# For anyOf and oneOf, the property_schema is a list of types.
137			if isinstance(property_type_schema, list):
138			return len([t for t in property_type_schema
139			if ((isinstance(t, six.string_types) and t == 'null') or
140			(isinstance(t, dict) and t.get('type', 'string') == 'null'))]) > 0
141
142			return (isinstance(property_type_schema, dict) and
143			property_type_schema.get('type', 'string') == 'null')
144
145
146			def is_attribute_type_array(attribute_type):
147			return (attribute_type == 'array' or
148			(isinstance(attribute_type, list) and 'array' in attribute_type))
149
150
151			def is_attribute_type_object(attribute_type):
152			return (attribute_type == 'object' or
153			(isinstance(attribute_type, list) and 'object' in attribute_type))
154
155
156			def assign_default_values(instance, schema):
157			"""
158			Assign default values on the provided instance based on the schema default specification.
159			"""
160			instance = copy.deepcopy(instance)
161			instance_is_dict = isinstance(instance, dict)
162			instance_is_array = isinstance(instance, list)
163
164			if not instance_is_dict and not instance_is_array:
165			return instance
166
167			properties = schema.get('properties', {})
168
169			for property_name, property_data in six.iteritems(properties):
170			has_default_value = 'default' in property_data
171			default_value = property_data.get('default', None)
172
173			# Assign default value on the instance so the validation doesn't fail if requires is true
174			# but the value is not provided
175			if has_default_value:
176			if instance_is_dict and instance.get(property_name, None) is None:
177			instance[property_name] = default_value
178			elif instance_is_array:
179			for index, _ in enumerate(instance):
180			if instance[index].get(property_name, None) is None:
181			instance[index][property_name] = default_value
182
183			# Support for nested properties (array and object)
184			attribute_type = property_data.get('type', None)
185			schema_items = property_data.get('items', {})
186
187			# Array
188			if (is_attribute_type_array(attribute_type) and
189			schema_items and schema_items.get('properties', {})):
190			array_instance = instance.get(property_name, None)
191			array_schema = schema['properties'][property_name]['items']
192
193			if array_instance is not None:
194			# Note: We don't perform subschema assignment if no value is provided
195			instance[property_name] = assign_default_values(instance=array_instance,
196			schema=array_schema)
197
198			# Object
199			if is_attribute_type_object(attribute_type) and property_data.get('properties', {}):
200			object_instance = instance.get(property_name, None)
201			object_schema = schema['properties'][property_name]
202
203			if object_instance is not None:
204			# Note: We don't perform subschema assignment if no value is provided
205			instance[property_name] = assign_default_values(instance=object_instance,
206			schema=object_schema)
207
208			return instance
209
210
211			def modify_schema_allow_default_none(schema):
212			"""
213			Manipulate the provided schema so None is also an allowed value for each attribute which
214			defines a default value of None.
215			"""
216			schema = copy.deepcopy(schema)
217			properties = schema.get('properties', {})
218
219			for property_name, property_data in six.iteritems(properties):
220			is_optional = not property_data.get('required', False)
221			has_default_value = 'default' in property_data
222			default_value = property_data.get('default', None)
223			property_schema = schema['properties'][property_name]
224
225			if (has_default_value or is_optional) and default_value is None:
226			# If property is anyOf and oneOf then it has to be process differently.
227			if (is_property_type_anyof(property_schema) and
228			not is_property_nullable(property_schema['anyOf'])):
229			property_schema['anyOf'].append({'type': 'null'})
230			elif (is_property_type_oneof(property_schema) and
231			not is_property_nullable(property_schema['oneOf'])):
232			property_schema['oneOf'].append({'type': 'null'})
233			elif (is_property_type_list(property_schema) and
234			not is_property_nullable(property_schema.get('type'))):
235			property_schema['type'].append('null')
236			elif (is_property_type_single(property_schema) and
237			not is_property_nullable(property_schema.get('type'))):
238			property_schema['type'] = [property_schema.get('type', 'string'), 'null']
239
240			# Support for nested properties (array and object)
241			attribute_type = property_data.get('type', None)
242			schema_items = property_data.get('items', {})
243
244			# Array
245			if (is_attribute_type_array(attribute_type) and
246			schema_items and schema_items.get('properties', {})):
247			array_schema = schema_items
248			array_schema = modify_schema_allow_default_none(schema=array_schema)
249			schema['properties'][property_name]['items'] = array_schema
250
251			# Object
252			if is_attribute_type_object(attribute_type) and property_data.get('properties', {}):
253			object_schema = property_data
254			object_schema = modify_schema_allow_default_none(schema=object_schema)
255			schema['properties'][property_name] = object_schema
256
257			return schema
258
259
260			def validate(instance, schema, cls=None, use_default=True, allow_default_none=False, *args,
261			**kwargs):
262			"""
263			Custom validate function which supports default arguments combined with the "required"
264			property.
265
266			Note: This function returns cleaned instance with default values assigned.
267
268			:param use_default: True to support the use of the optional "default" property.
269			:type use_default: ``bool``
270			"""
271			instance = copy.deepcopy(instance)
272			schema_type = schema.get('type', None)
273			instance_is_dict = isinstance(instance, dict)
274
275			if use_default and allow_default_none:
276			schema = modify_schema_allow_default_none(schema=schema)
277
278			if use_default and schema_type == 'object' and instance_is_dict:
279			instance = assign_default_values(instance=instance, schema=schema)
280
281			# pylint: disable=assignment-from-no-return
282			jsonschema.validate(instance=instance, schema=schema, cls=cls, args, *kwargs)
283
284			return instance
285
286
287			VALIDATORS = {
288			'draft4': jsonschema.Draft4Validator,
289			'custom': CustomValidator
290			}
291
292
293			def get_validator(version='custom'):
294			validator = VALIDATORS[version]
295			return validator
296
297
298			def get_schema_for_action_parameters(action_db):
299			"""
300			Dynamically construct JSON schema for the provided action from the parameters metadata.
301
302			Note: This schema is used to validate parameters which are passed to the action.
303			"""
304			from st2common.util.action_db import get_runnertype_by_name
305			runner_type = get_runnertype_by_name(action_db.runner_type['name'])
306
307			parameters_schema = {}
308			parameters_schema.update(runner_type.runner_parameters)
309			parameters_schema.update(action_db.parameters)
310
311			schema = get_schema_for_resource_parameters(parameters_schema=parameters_schema)
312
313			if parameters_schema:
314			schema['title'] = action_db.name
315			if action_db.description:
316			schema['description'] = action_db.description
317
318			return schema
319
320
321			def get_schema_for_resource_parameters(parameters_schema):
322			"""
323			Dynamically construct JSON schema for the provided resource from the parameters metadata.
324			"""
325			def normalize(x):
326			return {k: v if v else SCHEMA_ANY_TYPE for k, v in six.iteritems(x)}
327
328			schema = {}
329			properties = {}
330			properties.update(normalize(parameters_schema))
331			if properties:
332			schema['type'] = 'object'
333			schema['properties'] = properties
334			schema['additionalProperties'] = False
335
336			return schema
337

StackStorm / st2

Pull Request — master (#2335)

st2common.util.schema.is_property_type_list() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like