Source code for glance.schema

# Copyright 2012 OpenStack Foundation.
# All Rights Reserved.
#
#    Licensed 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 jsonschema
from oslo_utils import encodeutils

from glance.common import exception
from glance.i18n import _


[docs]class Schema(object): def __init__(self, name, properties=None, links=None, required=None, definitions=None): self.name = name if properties is None: properties = {} self.properties = properties self.links = links self.required = required self.definitions = definitions
[docs] def validate(self, obj): try: jsonschema.validate(obj, self.raw()) except jsonschema.ValidationError as e: reason = encodeutils.exception_to_unicode(e) raise exception.InvalidObject(schema=self.name, reason=reason)
[docs] def filter(self, obj): filtered = {} for key, value in obj.items(): if self._filter_func(self.properties, key): filtered[key] = value # NOTE(flaper87): This exists to allow for v1, null properties, # to be used with the V2 API. During Kilo, it was allowed for the # later to return None values without considering that V1 allowed # for custom properties to be None, which is something V2 doesn't # allow for. This small hack here will set V1 custom `None` pro- # perties to an empty string so that they will be updated along # with the image (if an update happens). # # We could skip the properties that are `None` but that would bring # back the behavior we moved away from. Note that we can't consider # doing a schema migration because we don't know which properties # are "custom" and which came from `schema-image` if those custom # properties were created with v1. if key not in self.properties and value is None: filtered[key] = '' return filtered
@staticmethod def _filter_func(properties, key): return key in properties
[docs] def merge_properties(self, properties): # Ensure custom props aren't attempting to override base props original_keys = set(self.properties.keys()) new_keys = set(properties.keys()) intersecting_keys = original_keys.intersection(new_keys) conflicting_keys = [k for k in intersecting_keys if self.properties[k] != properties[k]] if conflicting_keys: props = ', '.join(conflicting_keys) reason = _("custom properties (%(props)s) conflict " "with base properties") raise exception.SchemaLoadError(reason=reason % {'props': props}) self.properties.update(properties)
[docs] def raw(self): raw = { 'name': self.name, 'properties': self.properties, 'additionalProperties': False, } if self.definitions: raw['definitions'] = self.definitions if self.required: raw['required'] = self.required if self.links: raw['links'] = self.links return raw
[docs] def minimal(self): minimal = { 'name': self.name, 'properties': self.properties } if self.definitions: minimal['definitions'] = self.definitions if self.required: minimal['required'] = self.required return minimal
[docs]class PermissiveSchema(Schema): @staticmethod def _filter_func(properties, key): return True
[docs] def raw(self): raw = super(PermissiveSchema, self).raw() raw['additionalProperties'] = {'type': 'string'} return raw
[docs] def minimal(self): minimal = super(PermissiveSchema, self).raw() return minimal
[docs]class CollectionSchema(object): def __init__(self, name, item_schema): self.name = name self.item_schema = item_schema
[docs] def raw(self): definitions = None if self.item_schema.definitions: definitions = self.item_schema.definitions self.item_schema.definitions = None raw = { 'name': self.name, 'properties': { self.name: { 'type': 'array', 'items': self.item_schema.raw(), }, 'first': {'type': 'string'}, 'next': {'type': 'string'}, 'schema': {'type': 'string'}, }, 'links': [ {'rel': 'first', 'href': '{first}'}, {'rel': 'next', 'href': '{next}'}, {'rel': 'describedby', 'href': '{schema}'}, ], } if definitions: raw['definitions'] = definitions self.item_schema.definitions = definitions return raw
[docs] def minimal(self): definitions = None if self.item_schema.definitions: definitions = self.item_schema.definitions self.item_schema.definitions = None minimal = { 'name': self.name, 'properties': { self.name: { 'type': 'array', 'items': self.item_schema.minimal(), }, 'schema': {'type': 'string'}, }, 'links': [ {'rel': 'describedby', 'href': '{schema}'}, ], } if definitions: minimal['definitions'] = definitions self.item_schema.definitions = definitions return minimal
[docs]class DictCollectionSchema(Schema): def __init__(self, name, item_schema): self.name = name self.item_schema = item_schema
[docs] def raw(self): definitions = None if self.item_schema.definitions: definitions = self.item_schema.definitions self.item_schema.definitions = None raw = { 'name': self.name, 'properties': { self.name: { 'type': 'object', 'additionalProperties': self.item_schema.raw(), }, 'first': {'type': 'string'}, 'next': {'type': 'string'}, 'schema': {'type': 'string'}, }, 'links': [ {'rel': 'first', 'href': '{first}'}, {'rel': 'next', 'href': '{next}'}, {'rel': 'describedby', 'href': '{schema}'}, ], } if definitions: raw['definitions'] = definitions self.item_schema.definitions = definitions return raw
[docs] def minimal(self): definitions = None if self.item_schema.definitions: definitions = self.item_schema.definitions self.item_schema.definitions = None minimal = { 'name': self.name, 'properties': { self.name: { 'type': 'object', 'additionalProperties': self.item_schema.minimal(), }, 'schema': {'type': 'string'}, }, 'links': [ {'rel': 'describedby', 'href': '{schema}'}, ], } if definitions: minimal['definitions'] = definitions self.item_schema.definitions = definitions return minimal