¡@
OpenStack Study: utils.py**** CubicPower OpenStack Study **** def datetime_or_none(dt): """Validate a datetime or None value.""" if dt is None: return None elif isinstance(dt, datetime.datetime): if dt.utcoffset() is None: # NOTE(danms): Legacy objects from sqlalchemy are stored in UTC, # but are returned without a timezone attached. # As a transitional aid, assume a tz-naive object is in UTC. return dt.replace(tzinfo=iso8601.iso8601.Utc()) else: return dt raise ValueError('A datetime.datetime is required here') # NOTE(danms): Being tolerant of isotime strings here will help us # during our objects transition
**** CubicPower OpenStack Study **** def datetime_or_str_or_none(val): if isinstance(val, six.string_types): return timeutils.parse_isotime(val) return datetime_or_none(val)
**** CubicPower OpenStack Study **** def int_or_none(val): """Attempt to parse an integer value, or None.""" if val is None: return val else: return int(val)
**** CubicPower OpenStack Study **** def str_value(val): if val is None: raise ValueError(_('None is not valid here')) return unicode(val)
**** CubicPower OpenStack Study **** def str_or_none(val): """Attempt to stringify a value, or None.""" if val is None: return val else: return str_value(val)
**** CubicPower OpenStack Study **** def cstring(val): if val is None: raise ValueError(_('None is not valid here')) return str(val)
**** CubicPower OpenStack Study **** def ip_or_none(version): """Return a version-specific IP address validator.""" def validator(val, version=version): if val is None: return val else: return netaddr.IPAddress(val, version=version) return validator
**** CubicPower OpenStack Study **** def nested_object(objclass, none_ok=True): def validator(val, objclass=objclass): if none_ok and val is None: return val if isinstance(val, objclass): return val raise ValueError('An object of class %s is required here' % objclass) return validator
**** CubicPower OpenStack Study **** def network_model_or_none(val): """Validate/Convert to a network_model.NetworkInfo, or None.""" if val is None: return val if isinstance(val, network_model.NetworkInfo): return val return network_model.NetworkInfo.hydrate(val)
**** CubicPower OpenStack Study **** def list_of_strings_or_none(val): if val is None: return val if not isinstance(val, list): raise ValueError(_('A list of strings is required here')) if not all([isinstance(x, six.string_types) for x in val]): raise ValueError(_('Invalid values found in list ' '(strings are required)')) return val
**** CubicPower OpenStack Study **** def dict_of_strings_or_none(val): if val is None: return val if not isinstance(val, dict): try: val = dict(val.iteritems()) except Exception: raise ValueError(_('A dict of strings is required here')) if not all([isinstance(x, six.string_types) for x in val.keys()]): raise ValueError(_('Invalid keys found in dict ' '(strings are required)')) if not all([isinstance(x, six.string_types) for x in val.values()]): raise ValueError(_('Invalid values found in dict ' '(strings are required)')) return val
**** CubicPower OpenStack Study **** def dt_serializer(name): """Return a datetime serializer for a named attribute.""" def serializer(self, name=name): if getattr(self, name) is not None: return timeutils.isotime(getattr(self, name)) else: return None return serializer
**** CubicPower OpenStack Study **** def dt_deserializer(instance, val): """A deserializer method for datetime attributes.""" if val is None: return None else: return timeutils.parse_isotime(val)
**** CubicPower OpenStack Study **** def obj_serializer(name): def serializer(self, name=name): if getattr(self, name) is not None: return getattr(self, name).obj_to_primitive() else: return None return serializer
|