# Copyright (c) 2013 Spotify AB
#
# 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 datetime
import core
import copy
from core import ParseError, Field
import binascii
[docs]class Text(Field):
[docs] def load(self, obj):
if not isinstance(obj, (unicode, type(None))):
raise ParseError("%r not a unicode object" % obj)
return obj
[docs] def dump(self, obj):
if isinstance(obj, (unicode, type(None))):
return obj
else:
try:
return obj.decode('utf8')
except:
raise ValueError(
"%r is not a valid UTF-8 string" % obj
)
[docs]class Bytes(Field):
"""Binary data"""
def __init__(self, custom_encoding=False, **kwargs):
super(Bytes, self).__init__(**kwargs)
self.custom_encoding = custom_encoding
def _load_utf8_codepoints(self, obj):
return obj.encode("iso-8859-1")
def _dump_utf8_codepoints(self, binary_data):
return binary_data.decode("iso-8859-1")
def _load_b64(self, obj):
return binascii.a2b_base64(obj.encode("ascii"))
def _dump_b64(self, binary_data):
return binascii.b2a_base64(binary_data).rstrip('\n')
[docs] def load(self, obj):
if not self.custom_encoding:
return self._load_utf8_codepoints(obj)
return self._load_b64(obj)
[docs] def dump(self, binary_data):
if isinstance(binary_data, unicode):
raise ValueError(
"Unicode objects are not accepted values for Bytes (%r)"
% (binary_data,)
)
if not self.custom_encoding:
return self._dump_utf8_codepoints(binary_data)
return self._dump_b64(binary_data)
[docs]class List(Field):
"""List of one other Field type
Differs from other fields in that it is not nullable
and defaults to empty array instead of null
"""
def __init__(self, field_type=Text(), nullable=False, default=[], **kwargs):
super(List, self).__init__(nullable=nullable, default=default, **kwargs)
self.field_type = field_type
[docs] def load(self, obj):
if not isinstance(obj, list):
raise ParseError("%r is not a list object" % obj)
return [self.field_type.load(o) for o in obj]
[docs] def dump(self, obj):
if not isinstance(obj, (tuple, list)):
raise ValueError("%r is not a list object" % obj)
return [self.field_type.dump(o) for o in obj]
[docs] def set_parent(self, schema):
self.field_type.set_parent(schema)
[docs] def default_value(self):
# avoid default-sharing between records
return copy.deepcopy(self.default)
[docs]class Enum(Field):
_field_type = Text() # don't change
def __init__(self, values, **kwargs):
super(Enum, self).__init__(**kwargs)
self.values = set(values)
[docs] def dump(self, obj):
if obj not in self.values:
raise ValueError(
"%r is not an allowed value of Enum%r"
% (obj, tuple(self.values)))
return self._field_type.dump(obj)
[docs] def load(self, obj):
parsed = self._field_type.load(obj)
if parsed not in self.values and parsed is not None:
raise ParseError(
"Parsed value %r not in allowed value of Enum(%r)"
% (parsed, tuple(self.values)))
return parsed
[docs]class Integer(Field):
def __init__(self, size=8, **kwargs):
super(Integer, self).__init__(**kwargs)
self.size = size
[docs] def dump(self, obj):
if not isinstance(obj, (int, long, type(None))) or isinstance(obj, bool):
raise ValueError("%r is not a valid Integer" % (obj,))
return obj
[docs] def load(self, obj):
if not isinstance(obj, (int, long, type(None))) or isinstance(obj, bool):
raise ParseError("%r is not a valid Integer" % (obj,))
return obj
[docs]class Boolean(Field):
VALUE_MAP = {True: '1', 1: '1',
False: '0', 0: '0'}
[docs] def dump(self, obj):
if obj not in self.VALUE_MAP:
raise ValueError(
"Invalid value for Boolean field: %r" % obj)
return bool(obj)
[docs] def load(self, obj):
if obj not in self.VALUE_MAP:
raise ParseError(
"Invalid value for Boolean field: %r" % obj)
return bool(obj)
[docs]class Float(Field):
def __init__(self, size=8, **kwargs):
super(Float, self).__init__(**kwargs)
self.size = size
[docs] def dump(self, obj):
if not isinstance(obj, float):
raise ValueError("Invalid value for Float field: %r" % obj)
return float(obj)
[docs] def load(self, obj):
if not isinstance(obj, float):
raise ParseError("Invalid value for Float field: %r" % obj)
return float(obj)
[docs]class Date(Text):
[docs] def dump(self, obj):
if not isinstance(obj, datetime.date):
raise ValueError("Invalid value for Date field: %r" % obj)
return str(obj)
[docs] def load(self, obj):
try:
return datetime.datetime.strptime(obj, "%Y-%m-%d").date()
except ValueError:
raise ValueError("Invalid value for Date field: %r" % obj)
[docs]class DateTime(Text):
[docs] def dump(self, obj):
if not isinstance(obj, datetime.datetime):
raise ValueError("Invalid value for DateTime field: %r" % obj)
return str(obj)
[docs] def load(self, obj):
try:
if '.' in obj:
return datetime.datetime.strptime(obj, "%Y-%m-%d %H:%M:%S.%f")
return datetime.datetime.strptime(obj, "%Y-%m-%d %H:%M:%S")
except ValueError:
raise ValueError("Invalid value for DateField field: %r" % obj)
# special value for SubRecord's schema parameter
# that signifies a SubRecord accepting records of the
# same type as the container/parent Record.
SELF = object()
[docs]class SubRecord(Field):
""""Field for storing other :class:`record.Record`s"""
def __init__(self, schema, **kwargs):
super(SubRecord, self).__init__(**kwargs)
self._schema = schema
[docs] def dump(self, obj):
if not isinstance(obj, self._schema):
raise ValueError("%r is not a %r"
% (obj, self._schema))
return core.to_json_compatible(obj)
[docs] def load(self, obj):
return core.from_json_compatible(self._schema, obj)
[docs] def set_parent(self, schema):
"""This method gets called by the metaclass
once the container class has been created
to let the field store a reference to its
parent if needed. Its needed for SubRecords
in case it refers to the container record.
"""
if self._schema == SELF:
self._schema = schema
[docs] def default_value(self):
# avoid default-sharing between records
return copy.deepcopy(self.default)
[docs]class Map(Field):
"""List of one other Field type
Differs from other fields in that it is not nullable
and defaults to empty array instead of null
"""
def __init__(self, value_type, nullable=False, default={}, **kwargs):
super(Map, self).__init__(nullable=nullable, default=default, **kwargs)
self.value_type = value_type
self.key_type = Text()
[docs] def load(self, obj):
return dict([
(self.key_type.load(k),
self.value_type.load(v))
for k, v in obj.iteritems()
])
[docs] def dump(self, obj):
if not isinstance(obj, dict):
raise ValueError("%r is not a dict" % (obj,))
return dict([
(self.key_type.dump(k),
self.value_type.dump(v))
for k, v in obj.iteritems()
])
[docs] def set_parent(self, schema):
self.value_type.set_parent(schema)
[docs] def default_value(self):
# avoid default-sharing between records
return copy.deepcopy(self.default)