¡@
OpenStack Study: fields.py**** CubicPower OpenStack Study ****
# Copyright 2013 Red Hat, Inc. # # 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 abc import datetime import iso8601 import netaddr import six from nova.network import model as network_model from nova.openstack.common.gettextutils import _ from nova.openstack.common import timeutils
**** CubicPower OpenStack Study **** class KeyTypeError(TypeError):
**** CubicPower OpenStack Study **** def __init__(self, expected, value): super(KeyTypeError, self).__init__( _('Key %(key)s must be of type %(expected)s not %(actual)s' ) % {'key': repr(value), 'expected': expected.__name__, 'actual': value.__class__.__name__, })
**** CubicPower OpenStack Study **** class ElementTypeError(TypeError):
**** CubicPower OpenStack Study **** def __init__(self, expected, key, value): super(ElementTypeError, self).__init__( _('Element %(key)s:%(val)s must be of type %(expected)s' ' not %(actual)s' ) % {'key': key, 'val': repr(value), 'expected': expected, 'actual': value.__class__.__name__, })
**** CubicPower OpenStack Study **** class AbstractFieldType(six.with_metaclass(abc.ABCMeta, object)): @abc.abstractmethod
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): """This is called to coerce (if possible) a value on assignment. This method should convert the value given into the designated type, or throw an exception if this is not possible. :param:obj: The NovaObject on which an attribute is being set :param:attr: The name of the attribute being set :param:value: The value being set :returns: A properly-typed value """ pass @abc.abstractmethod
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): """This is called to deserialize a value. This method should deserialize a value from the form given by to_primitive() to the designated type. :param:obj: The NovaObject on which the value is to be set :param:attr: The name of the attribute which will hold the value :param:value: The serialized form of the value :returns: The natural form of the value """ pass @abc.abstractmethod
**** CubicPower OpenStack Study **** def to_primitive(self, obj, attr, value): """This is called to serialize a value. This method should serialize a value to the form expected by from_primitive(). :param:obj: The NovaObject on which the value is set :param:attr: The name of the attribute holding the value :param:value: The natural form of the value :returns: The serialized form of the value """ pass @abc.abstractmethod
**** CubicPower OpenStack Study **** def describe(self): """Returns a string describing the type of the field.""" pass
**** CubicPower OpenStack Study **** class FieldType(AbstractFieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): return value @staticmethod
**** CubicPower OpenStack Study **** def from_primitive(obj, attr, value): return value @staticmethod
**** CubicPower OpenStack Study **** def to_primitive(obj, attr, value): return value
**** CubicPower OpenStack Study **** def describe(self): return self.__class__.__name__
**** CubicPower OpenStack Study **** class UnspecifiedDefault(object): pass
**** CubicPower OpenStack Study **** class Field(object):
**** CubicPower OpenStack Study **** def __init__(self, field_type, nullable=False, default=UnspecifiedDefault): self._type = field_type self._nullable = nullable self._default = default @property
**** CubicPower OpenStack Study **** def nullable(self): return self._nullable @property
**** CubicPower OpenStack Study **** def default(self): return self._default
**** CubicPower OpenStack Study **** def _null(self, obj, attr): if self.nullable: return None elif self._default != UnspecifiedDefault: # NOTE(danms): We coerce the default value each time the field # is set to None as our contract states that we'll let the type # examine the object and attribute name at that time. return self._type.coerce(obj, attr, self._default) else: raise ValueError(_("Field `%s' cannot be None") % attr)
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): """Coerce a value to a suitable type. This is called any time you set a value on an object, like: foo.myint = 1 and is responsible for making sure that the value (1 here) is of the proper type, or can be sanely converted. This also handles the potentially nullable or defaultable nature of the field and calls the coerce() method on a FieldType to actually do the coercion. :param:obj: The object being acted upon :param:attr: The name of the attribute/field being set :param:value: The value being set :returns: The properly-typed value """ if value is None: return self._null(obj, attr) else: return self._type.coerce(obj, attr, value)
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): """Deserialize a value from primitive form. This is responsible for deserializing a value from primitive into regular form. It calls the from_primitive() method on a FieldType to do the actual deserialization. :param:obj: The object being acted upon :param:attr: The name of the attribute/field being deserialized :param:value: The value to be deserialized :returns: The deserialized value """ if value is None: return None else: return self._type.from_primitive(obj, attr, value)
**** CubicPower OpenStack Study **** def to_primitive(self, obj, attr, value): """Serialize a value to primitive form. This is responsible for serializing a value to primitive form. It calls to_primitive() on a FieldType to do the actual serialization. :param:obj: The object being acted upon :param:attr: The name of the attribute/field being serialized :param:value: The value to be serialized :returns: The serialized value """ if value is None: return None else: return self._type.to_primitive(obj, attr, value)
**** CubicPower OpenStack Study **** def describe(self): """Return a short string describing the type of this field.""" name = self._type.describe() prefix = self.nullable and 'Nullable' or '' return prefix + name
**** CubicPower OpenStack Study **** class String(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): # FIXME(danms): We should really try to avoid the need to do this if isinstance(value, (six.string_types, int, long, float, datetime.datetime)): return unicode(value) else: raise ValueError(_('A string is required here, not %s') % value.__class__.__name__)
**** CubicPower OpenStack Study **** class UUID(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): # FIXME(danms): We should actually verify the UUIDness here return str(value)
**** CubicPower OpenStack Study **** class Integer(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): return int(value)
**** CubicPower OpenStack Study **** class Float(FieldType):
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): return float(value)
**** CubicPower OpenStack Study **** class Boolean(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): return bool(value)
**** CubicPower OpenStack Study **** class DateTime(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): if isinstance(value, six.string_types): value = timeutils.parse_isotime(value) elif not isinstance(value, datetime.datetime): raise ValueError(_('A datetime.datetime is required here')) if value.utcoffset() is None: value = value.replace(tzinfo=iso8601.iso8601.Utc()) return value
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): return self.coerce(obj, attr, timeutils.parse_isotime(value)) @staticmethod
**** CubicPower OpenStack Study **** def to_primitive(obj, attr, value): return timeutils.isotime(value)
**** CubicPower OpenStack Study **** class IPAddress(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): try: return netaddr.IPAddress(value) except netaddr.AddrFormatError as e: raise ValueError(str(e))
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): return self.coerce(obj, attr, value) @staticmethod
**** CubicPower OpenStack Study **** def to_primitive(obj, attr, value): return str(value)
**** CubicPower OpenStack Study **** class IPV4Address(IPAddress): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): result = IPAddress.coerce(obj, attr, value) if result.version != 4: raise ValueError(_('Network "%s" is not valid') % value) return result
**** CubicPower OpenStack Study **** class IPV6Address(IPAddress): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): result = IPAddress.coerce(obj, attr, value) if result.version != 6: raise ValueError(_('Network "%s" is not valid') % value) return result
**** CubicPower OpenStack Study **** class IPV4AndV6Address(IPAddress): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): result = IPAddress.coerce(obj, attr, value) if result.version != 4 and result.version != 6: raise ValueError(_('Network "%s" is not valid') % value) return result
**** CubicPower OpenStack Study **** class IPNetwork(IPAddress): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): try: return netaddr.IPNetwork(value) except netaddr.AddrFormatError as e: raise ValueError(str(e))
**** CubicPower OpenStack Study **** class IPV4Network(IPNetwork): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): try: return netaddr.IPNetwork(value, version=4) except netaddr.AddrFormatError as e: raise ValueError(str(e))
**** CubicPower OpenStack Study **** class IPV6Network(IPNetwork): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): try: return netaddr.IPNetwork(value, version=6) except netaddr.AddrFormatError as e: raise ValueError(str(e))
**** CubicPower OpenStack Study **** class CompoundFieldType(FieldType):
**** CubicPower OpenStack Study **** def __init__(self, element_type, **field_args): self._element_type = Field(element_type, **field_args)
**** CubicPower OpenStack Study **** class List(CompoundFieldType):
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): if not isinstance(value, list): raise ValueError(_('A list is required here')) for index, element in enumerate(list(value)): value[index] = self._element_type.coerce( obj, '%s[%i]' % (attr, index), element) return value
**** CubicPower OpenStack Study **** def to_primitive(self, obj, attr, value): return [self._element_type.to_primitive(obj, attr, x) for x in value]
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): return [self._element_type.from_primitive(obj, attr, x) for x in value]
**** CubicPower OpenStack Study **** class Dict(CompoundFieldType):
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): if not isinstance(value, dict): raise ValueError(_('A dict is required here')) for key, element in value.items(): if not isinstance(key, six.string_types): #NOTE(guohliu) In order to keep compatibility with python3 #we need to use six.string_types rather than basestring here, #since six.string_types is a tuple, so we need to pass the #real type in. raise KeyTypeError(six.string_types[0], key) value[key] = self._element_type.coerce( obj, '%s["%s"]' % (attr, key), element) return value
**** CubicPower OpenStack Study **** def to_primitive(self, obj, attr, value): primitive = {} for key, element in value.items(): primitive[key] = self._element_type.to_primitive( obj, '%s["%s"]' % (attr, key), element) return primitive
**** CubicPower OpenStack Study **** def from_primitive(self, obj, attr, value): concrete = {} for key, element in value.items(): concrete[key] = self._element_type.from_primitive( obj, '%s["%s"]' % (attr, key), element) return concrete
**** CubicPower OpenStack Study **** class Object(FieldType):
**** CubicPower OpenStack Study **** def __init__(self, obj_name, **kwargs): self._obj_name = obj_name super(Object, self).__init__(**kwargs)
**** CubicPower OpenStack Study **** def coerce(self, obj, attr, value): try: obj_name = value.obj_name() except AttributeError: obj_name = "" if obj_name != self._obj_name: raise ValueError(_('An object of type %s is required here') % self._obj_name) return value @staticmethod
**** CubicPower OpenStack Study **** def to_primitive(obj, attr, value): return value.obj_to_primitive() @staticmethod
**** CubicPower OpenStack Study **** def from_primitive(obj, attr, value): # FIXME(danms): Avoid circular import from base.py from nova.objects import base as obj_base return obj_base.NovaObject.obj_from_primitive(value, obj._context)
**** CubicPower OpenStack Study **** def describe(self): return "Object<%s>" % self._obj_name
**** CubicPower OpenStack Study **** class NetworkModel(FieldType): @staticmethod
**** CubicPower OpenStack Study **** def coerce(obj, attr, value): if isinstance(value, network_model.NetworkInfo): return value elif isinstance(value, six.string_types): # Hmm, do we need this? return network_model.NetworkInfo.hydrate(value) else: raise ValueError(_('A NetworkModel is required here')) @staticmethod
**** CubicPower OpenStack Study **** def to_primitive(obj, attr, value): return value.json() @staticmethod
**** CubicPower OpenStack Study **** def from_primitive(obj, attr, value): return network_model.NetworkInfo.hydrate(value)
**** CubicPower OpenStack Study **** class AutoTypedField(Field): AUTO_TYPE = None
**** CubicPower OpenStack Study **** def __init__(self, **kwargs): super(AutoTypedField, self).__init__(self.AUTO_TYPE, **kwargs)
**** CubicPower OpenStack Study **** class StringField(AutoTypedField): AUTO_TYPE = String()
**** CubicPower OpenStack Study **** class UUIDField(AutoTypedField): AUTO_TYPE = UUID()
**** CubicPower OpenStack Study **** class IntegerField(AutoTypedField): AUTO_TYPE = Integer()
**** CubicPower OpenStack Study **** class FloatField(AutoTypedField): AUTO_TYPE = Float()
**** CubicPower OpenStack Study **** class BooleanField(AutoTypedField): AUTO_TYPE = Boolean()
**** CubicPower OpenStack Study **** class DateTimeField(AutoTypedField): AUTO_TYPE = DateTime()
**** CubicPower OpenStack Study **** class IPAddressField(AutoTypedField): AUTO_TYPE = IPAddress()
**** CubicPower OpenStack Study **** class IPV4AddressField(AutoTypedField): AUTO_TYPE = IPV4Address()
**** CubicPower OpenStack Study **** class IPV6AddressField(AutoTypedField): AUTO_TYPE = IPV6Address()
**** CubicPower OpenStack Study **** class IPNetworkField(AutoTypedField): AUTO_TYPE = IPNetwork()
**** CubicPower OpenStack Study **** class IPV4NetworkField(AutoTypedField): AUTO_TYPE = IPV4Network()
**** CubicPower OpenStack Study **** class IPV6NetworkField(AutoTypedField): AUTO_TYPE = IPV6Network()
**** CubicPower OpenStack Study **** class DictOfStringsField(AutoTypedField): AUTO_TYPE = Dict(String())
**** CubicPower OpenStack Study **** class DictOfNullableStringsField(AutoTypedField): AUTO_TYPE = Dict(String(), nullable=True)
**** CubicPower OpenStack Study **** class ListOfStringsField(AutoTypedField): AUTO_TYPE = List(String())
**** CubicPower OpenStack Study **** class ObjectField(AutoTypedField):
**** CubicPower OpenStack Study **** def __init__(self, objtype, **kwargs): self.AUTO_TYPE = Object(objtype) super(ObjectField, self).__init__(**kwargs)
**** CubicPower OpenStack Study **** class ListOfObjectsField(AutoTypedField):
**** CubicPower OpenStack Study **** def __init__(self, objtype, **kwargs): self.AUTO_TYPE = List(Object(objtype)) super(ListOfObjectsField, self).__init__(**kwargs)
|