#!/usr/bin/env python

##
# Licensed to the Apache Software Foundation (ASF) 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
#
# https://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.

"""
Input/Output utilities, including:

 * i/o-specific constants
 * i/o-specific exceptions
 * schema validation
 * leaf value encoding and decoding
 * datum reader/writer stuff (?)

Also includes a generic representation for data, which
uses the following mapping:

  * Schema records are implemented as dict.
  * Schema arrays are implemented as list.
  * Schema maps are implemented as dict.
  * Schema strings are implemented as unicode.
  * Schema bytes are implemented as str.
  * Schema ints are implemented as int.
  * Schema longs are implemented as long.
  * Schema floats are implemented as float.
  * Schema doubles are implemented as float.
  * Schema booleans are implemented as bool.
"""

from __future__ import absolute_import, division, print_function

import datetime
import json
import struct
import sys
from decimal import Decimal, getcontext
from struct import Struct

from avro import constants, schema, timezones

try:
    unicode
except NameError:
    unicode = str

try:
    basestring  # type: ignore
except NameError:
    basestring = (bytes, unicode)

try:
    long
except NameError:
    long = int


#
# Constants
#

_DEBUG_VALIDATE_INDENT = 0
_DEBUG_VALIDATE = False

INT_MIN_VALUE = -(1 << 31)
INT_MAX_VALUE = (1 << 31) - 1
LONG_MIN_VALUE = -(1 << 63)
LONG_MAX_VALUE = (1 << 63) - 1

# TODO(hammer): shouldn't ! be < for little-endian (according to spec?)
STRUCT_FLOAT = Struct('<f')           # big-endian float
STRUCT_DOUBLE = Struct('<d')          # big-endian double
STRUCT_SIGNED_SHORT = Struct('>h')    # big-endian signed short
STRUCT_SIGNED_INT = Struct('>i')      # big-endian signed int
STRUCT_SIGNED_LONG = Struct('>q')     # big-endian signed long


#
# Exceptions
#

class AvroTypeException(schema.AvroException):
    """Raised when datum is not an example of schema."""

    def __init__(self, expected_schema, datum):
        pretty_expected = json.dumps(json.loads(str(expected_schema)), indent=2)
        fail_msg = "The datum %s is not an example of the schema %s"\
                   % (datum, pretty_expected)
        schema.AvroException.__init__(self, fail_msg)


class SchemaResolutionException(schema.AvroException):
    def __init__(self, fail_msg, writers_schema=None, readers_schema=None):
        pretty_writers = json.dumps(json.loads(str(writers_schema)), indent=2)
        pretty_readers = json.dumps(json.loads(str(readers_schema)), indent=2)
        if writers_schema:
            fail_msg += "\nWriter's Schema: %s" % pretty_writers
        if readers_schema:
            fail_msg += "\nReader's Schema: %s" % pretty_readers
        schema.AvroException.__init__(self, fail_msg)

#
# Validate
#


def _is_timezone_aware_datetime(dt):
    return dt.tzinfo is not None and dt.tzinfo.utcoffset(dt) is not None


_valid = {
    'null': lambda s, d: d is None,
    'boolean': lambda s, d: isinstance(d, bool),
    'string': lambda s, d: isinstance(d, unicode),
    'bytes': lambda s, d: ((isinstance(d, bytes)) or
                           (isinstance(d, Decimal) and
                            getattr(s, 'logical_type', None) == constants.DECIMAL)),
    'int': lambda s, d: ((isinstance(d, (int, long))) and (INT_MIN_VALUE <= d <= INT_MAX_VALUE) or
                         (isinstance(d, datetime.date) and
                          getattr(s, 'logical_type', None) == constants.DATE) or
                         (isinstance(d, datetime.time) and
                          getattr(s, 'logical_type', None) == constants.TIME_MILLIS)),
    'long': lambda s, d: ((isinstance(d, (int, long))) and (LONG_MIN_VALUE <= d <= LONG_MAX_VALUE) or
                          (isinstance(d, datetime.time) and
                           getattr(s, 'logical_type', None) == constants.TIME_MICROS) or
                          (isinstance(d, datetime.date) and
                           _is_timezone_aware_datetime(d) and
                           getattr(s, 'logical_type', None) in (constants.TIMESTAMP_MILLIS,
                                                                constants.TIMESTAMP_MICROS))),
    'float': lambda s, d: isinstance(d, (int, long, float)),
    'fixed': lambda s, d: ((isinstance(d, bytes) and len(d) == s.size) or
                           (isinstance(d, Decimal) and
                            getattr(s, 'logical_type', None) == constants.DECIMAL)),
    'enum': lambda s, d: d in s.symbols,

    'array': lambda s, d: isinstance(d, list) and all(validate(s.items, item) for item in d),
    'map': lambda s, d: (isinstance(d, dict) and all(isinstance(key, unicode) for key in d) and
                         all(validate(s.values, value) for value in d.values())),
    'union': lambda s, d: any(validate(branch, d) for branch in s.schemas),
    'record': lambda s, d: (isinstance(d, dict) and
                            all(validate(f.type, d.get(f.name)) for f in s.fields) and
                            {f.name for f in s.fields}.issuperset(d.keys())),
}
_valid['double'] = _valid['float']
_valid['error_union'] = _valid['union']
_valid['error'] = _valid['request'] = _valid['record']


def validate(expected_schema, datum):
    """Determines if a python datum is an instance of a schema.

    Args:
      expected_schema: Schema to validate against.
      datum: Datum to validate.
    Returns:
      True if the datum is an instance of the schema.
    """
    global _DEBUG_VALIDATE_INDENT
    global _DEBUG_VALIDATE
    expected_type = expected_schema.type
    name = getattr(expected_schema, 'name', '')
    if name:
        name = ' ' + name
    if expected_type in ('array', 'map', 'union', 'record'):
        if _DEBUG_VALIDATE:
            print('{!s}{!s}{!s}: {!s} {{'.format(' ' * _DEBUG_VALIDATE_INDENT, expected_schema.type, name, type(datum).__name__), file=sys.stderr)
            _DEBUG_VALIDATE_INDENT += 2
            if datum is not None and not datum:
                print('{!s}<Empty>'.format(' ' * _DEBUG_VALIDATE_INDENT), file=sys.stderr)
        result = _valid[expected_type](expected_schema, datum)
        if _DEBUG_VALIDATE:
            _DEBUG_VALIDATE_INDENT -= 2
            print('{!s}}} -> {!s}'.format(' ' * _DEBUG_VALIDATE_INDENT, result), file=sys.stderr)
    else:
        result = _valid[expected_type](expected_schema, datum)
        if _DEBUG_VALIDATE:
            print('{!s}{!s}{!s}: {!s} -> {!s}'.format(' ' * _DEBUG_VALIDATE_INDENT,
                  expected_schema.type, name, type(datum).__name__, result), file=sys.stderr)
    return result


#
# Decoder/Encoder
#

class BinaryDecoder(object):
    """Read leaf values."""

    def __init__(self, reader):
        """
        reader is a Python object on which we can call read, seek, and tell.
        """
        self._reader = reader

    # read-only properties
    reader = property(lambda self: self._reader)

    def read(self, n):
        """
        Read n bytes.
        """
        return self.reader.read(n)

    def read_null(self):
        """
        null is written as zero bytes
        """
        return None

    def read_boolean(self):
        """
        a boolean is written as a single byte
        whose value is either 0 (false) or 1 (true).
        """
        return ord(self.read(1)) == 1

    def read_int(self):
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        return self.read_long()

    def read_long(self):
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        b = ord(self.read(1))
        n = b & 0x7F
        shift = 7
        while (b & 0x80) != 0:
            b = ord(self.read(1))
            n |= (b & 0x7F) << shift
            shift += 7
        datum = (n >> 1) ^ -(n & 1)
        return datum

    def read_float(self):
        """
        A float is written as 4 bytes.
        The float is converted into a 32-bit integer using a method equivalent to
        Java's floatToIntBits and then encoded in little-endian format.
        """
        return STRUCT_FLOAT.unpack(self.read(4))[0]

    def read_double(self):
        """
        A double is written as 8 bytes.
        The double is converted into a 64-bit integer using a method equivalent to
        Java's doubleToLongBits and then encoded in little-endian format.
        """
        return STRUCT_DOUBLE.unpack(self.read(8))[0]

    def read_decimal_from_bytes(self, precision, scale):
        """
        Decimal bytes are decoded as signed short, int or long depending on the
        size of bytes.
        """
        size = self.read_long()
        return self.read_decimal_from_fixed(precision, scale, size)

    def read_decimal_from_fixed(self, precision, scale, size):
        """
        Decimal is encoded as fixed. Fixed instances are encoded using the
        number of bytes declared in the schema.
        """
        datum = self.read(size)
        unscaled_datum = 0
        msb = struct.unpack('!b', datum[0:1])[0]
        leftmost_bit = (msb >> 7) & 1
        if leftmost_bit == 1:
            modified_first_byte = ord(datum[0:1]) ^ (1 << 7)
            datum = bytearray([modified_first_byte]) + datum[1:]
            for offset in range(size):
                unscaled_datum <<= 8
                unscaled_datum += ord(datum[offset:1 + offset])
            unscaled_datum += pow(-2, (size * 8) - 1)
        else:
            for offset in range(size):
                unscaled_datum <<= 8
                unscaled_datum += ord(datum[offset:1 + offset])

        original_prec = getcontext().prec
        getcontext().prec = precision
        scaled_datum = Decimal(unscaled_datum).scaleb(-scale)
        getcontext().prec = original_prec
        return scaled_datum

    def read_bytes(self):
        """
        Bytes are encoded as a long followed by that many bytes of data.
        """
        return self.read(self.read_long())

    def read_utf8(self):
        """
        A string is encoded as a long followed by
        that many bytes of UTF-8 encoded character data.
        """
        return unicode(self.read_bytes(), "utf-8")

    def read_date_from_int(self):
        """
        int is decoded as python date object.
        int stores the number of days from
        the unix epoch, 1 January 1970 (ISO calendar).
        """
        days_since_epoch = self.read_int()
        return datetime.date(1970, 1, 1) + datetime.timedelta(days_since_epoch)

    def _build_time_object(self, value, scale_to_micro):
        value = value * scale_to_micro
        value, microseconds = value // 1000000, value % 1000000
        value, seconds = value // 60, value % 60
        value, minutes = value // 60, value % 60
        hours = value

        return datetime.time(
            hour=hours,
            minute=minutes,
            second=seconds,
            microsecond=microseconds
        )

    def read_time_millis_from_int(self):
        """
        int is decoded as python time object which represents
        the number of milliseconds after midnight, 00:00:00.000.
        """
        milliseconds = self.read_int()
        return self._build_time_object(milliseconds, 1000)

    def read_time_micros_from_long(self):
        """
        long is decoded as python time object which represents
        the number of microseconds after midnight, 00:00:00.000000.
        """
        microseconds = self.read_long()
        return self._build_time_object(microseconds, 1)

    def read_timestamp_millis_from_long(self):
        """
        long is decoded as python datetime object which represents
        the number of milliseconds from the unix epoch, 1 January 1970.
        """
        timestamp_millis = self.read_long()
        timedelta = datetime.timedelta(microseconds=timestamp_millis * 1000)
        unix_epoch_datetime = datetime.datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=timezones.utc)
        return unix_epoch_datetime + timedelta

    def read_timestamp_micros_from_long(self):
        """
        long is decoded as python datetime object which represents
        the number of microseconds from the unix epoch, 1 January 1970.
        """
        timestamp_micros = self.read_long()
        timedelta = datetime.timedelta(microseconds=timestamp_micros)
        unix_epoch_datetime = datetime.datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=timezones.utc)
        return unix_epoch_datetime + timedelta

    def skip_null(self):
        pass

    def skip_boolean(self):
        self.skip(1)

    def skip_int(self):
        self.skip_long()

    def skip_long(self):
        b = ord(self.read(1))
        while (b & 0x80) != 0:
            b = ord(self.read(1))

    def skip_float(self):
        self.skip(4)

    def skip_double(self):
        self.skip(8)

    def skip_bytes(self):
        self.skip(self.read_long())

    def skip_utf8(self):
        self.skip_bytes()

    def skip(self, n):
        self.reader.seek(self.reader.tell() + n)


class BinaryEncoder(object):
    """Write leaf values."""

    def __init__(self, writer):
        """
        writer is a Python object on which we can call write.
        """
        self._writer = writer

    # read-only properties
    writer = property(lambda self: self._writer)

    def write(self, datum):
        """Write an arbitrary datum."""
        self.writer.write(datum)

    def write_null(self, datum):
        """
        null is written as zero bytes
        """
        pass

    def write_boolean(self, datum):
        """
        a boolean is written as a single byte
        whose value is either 0 (false) or 1 (true).
        """
        self.write(bytearray([bool(datum)]))

    def write_int(self, datum):
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        self.write_long(datum)

    def write_long(self, datum):
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        datum = (datum << 1) ^ (datum >> 63)
        while (datum & ~0x7F) != 0:
            self.write(bytearray([(datum & 0x7f) | 0x80]))
            datum >>= 7
        self.write(bytearray([datum]))

    def write_float(self, datum):
        """
        A float is written as 4 bytes.
        The float is converted into a 32-bit integer using a method equivalent to
        Java's floatToIntBits and then encoded in little-endian format.
        """
        self.write(STRUCT_FLOAT.pack(datum))

    def write_double(self, datum):
        """
        A double is written as 8 bytes.
        The double is converted into a 64-bit integer using a method equivalent to
        Java's doubleToLongBits and then encoded in little-endian format.
        """
        self.write(STRUCT_DOUBLE.pack(datum))

    def write_decimal_bytes(self, datum, scale):
        """
        Decimal in bytes are encoded as long. Since size of packed value in bytes for
        signed long is 8, 8 bytes are written.
        """
        sign, digits, exp = datum.as_tuple()
        if exp > scale:
            raise AvroTypeException('Scale provided in schema does not match the decimal')

        unscaled_datum = 0
        for digit in digits:
            unscaled_datum = (unscaled_datum * 10) + digit

        bits_req = unscaled_datum.bit_length() + 1
        if sign:
            unscaled_datum = (1 << bits_req) - unscaled_datum

        bytes_req = bits_req // 8
        padding_bits = ~((1 << bits_req) - 1) if sign else 0
        packed_bits = padding_bits | unscaled_datum

        bytes_req += 1 if (bytes_req << 3) < bits_req else 0
        self.write_long(bytes_req)
        for index in range(bytes_req - 1, -1, -1):
            bits_to_write = packed_bits >> (8 * index)
            self.write(bytearray([bits_to_write & 0xff]))

    def write_decimal_fixed(self, datum, scale, size):
        """
        Decimal in fixed are encoded as size of fixed bytes.
        """
        sign, digits, exp = datum.as_tuple()
        if exp > scale:
            raise AvroTypeException('Scale provided in schema does not match the decimal')

        unscaled_datum = 0
        for digit in digits:
            unscaled_datum = (unscaled_datum * 10) + digit

        bits_req = unscaled_datum.bit_length() + 1
        size_in_bits = size * 8
        offset_bits = size_in_bits - bits_req

        mask = 2 ** size_in_bits - 1
        bit = 1
        for i in range(bits_req):
            mask ^= bit
            bit <<= 1

        if bits_req < 8:
            bytes_req = 1
        else:
            bytes_req = bits_req // 8
            if bits_req % 8 != 0:
                bytes_req += 1
        if sign:
            unscaled_datum = (1 << bits_req) - unscaled_datum
            unscaled_datum = mask | unscaled_datum
            for index in range(size - 1, -1, -1):
                bits_to_write = unscaled_datum >> (8 * index)
                self.write(bytearray([bits_to_write & 0xff]))
        else:
            for i in range(offset_bits // 8):
                self.write(b'\x00')
            for index in range(bytes_req - 1, -1, -1):
                bits_to_write = unscaled_datum >> (8 * index)
                self.write(bytearray([bits_to_write & 0xff]))

    def write_bytes(self, datum):
        """
        Bytes are encoded as a long followed by that many bytes of data.
        """
        self.write_long(len(datum))
        self.write(struct.pack('%ds' % len(datum), datum))

    def write_utf8(self, datum):
        """
        A string is encoded as a long followed by
        that many bytes of UTF-8 encoded character data.
        """
        datum = datum.encode("utf-8")
        self.write_bytes(datum)

    def write_date_int(self, datum):
        """
        Encode python date object as int.
        It stores the number of days from
        the unix epoch, 1 January 1970 (ISO calendar).
        """
        delta_date = datum - datetime.date(1970, 1, 1)
        self.write_int(delta_date.days)

    def write_time_millis_int(self, datum):
        """
        Encode python time object as int.
        It stores the number of milliseconds from midnight, 00:00:00.000
        """
        milliseconds = datum.hour * 3600000 + datum.minute * 60000 + datum.second * 1000 + datum.microsecond // 1000
        self.write_int(milliseconds)

    def write_time_micros_long(self, datum):
        """
        Encode python time object as long.
        It stores the number of microseconds from midnight, 00:00:00.000000
        """
        microseconds = datum.hour * 3600000000 + datum.minute * 60000000 + datum.second * 1000000 + datum.microsecond
        self.write_long(microseconds)

    def _timedelta_total_microseconds(self, timedelta):
        return (
            timedelta.microseconds + (timedelta.seconds + timedelta.days * 24 * 3600) * 10 ** 6)

    def write_timestamp_millis_long(self, datum):
        """
        Encode python datetime object as long.
        It stores the number of milliseconds from midnight of unix epoch, 1 January 1970.
        """
        datum = datum.astimezone(tz=timezones.utc)
        timedelta = datum - datetime.datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=timezones.utc)
        milliseconds = self._timedelta_total_microseconds(timedelta) / 1000
        self.write_long(long(milliseconds))

    def write_timestamp_micros_long(self, datum):
        """
        Encode python datetime object as long.
        It stores the number of microseconds from midnight of unix epoch, 1 January 1970.
        """
        datum = datum.astimezone(tz=timezones.utc)
        timedelta = datum - datetime.datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=timezones.utc)
        microseconds = self._timedelta_total_microseconds(timedelta)
        self.write_long(long(microseconds))


#
# DatumReader/Writer
#
class DatumReader(object):
    """Deserialize Avro-encoded data into a Python data structure."""

    def __init__(self, writers_schema=None, readers_schema=None):
        """
        As defined in the Avro specification, we call the schema encoded
        in the data the "writer's schema", and the schema expected by the
        reader the "reader's schema".
        """
        self._writers_schema = writers_schema
        self._readers_schema = readers_schema

    # read/write properties
    def set_writers_schema(self, writers_schema):
        self._writers_schema = writers_schema
    writers_schema = property(lambda self: self._writers_schema,
                              set_writers_schema)

    def set_readers_schema(self, readers_schema):
        self._readers_schema = readers_schema
    readers_schema = property(lambda self: self._readers_schema,
                              set_readers_schema)

    def read(self, decoder):
        if self.readers_schema is None:
            self.readers_schema = self.writers_schema
        return self.read_data(self.writers_schema, self.readers_schema, decoder)

    def read_data(self, writers_schema, readers_schema, decoder):
        # schema matching
        if not readers_schema.match(writers_schema):
            fail_msg = 'Schemas do not match.'
            raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)

        logical_type = getattr(writers_schema, 'logical_type', None)

        # function dispatch for reading data based on type of writer's schema
        if writers_schema.type in ['union', 'error_union']:
            return self.read_union(writers_schema, readers_schema, decoder)

        if readers_schema.type in ['union', 'error_union']:
            # schema resolution: reader's schema is a union, writer's schema is not
            for s in readers_schema.schemas:
                if s.match(writers_schema):
                    return self.read_data(writers_schema, s, decoder)

            # This shouldn't happen because of the match check at the start of this method.
            fail_msg = 'Schemas do not match.'
            raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)

        if writers_schema.type == 'null':
            return decoder.read_null()
        elif writers_schema.type == 'boolean':
            return decoder.read_boolean()
        elif writers_schema.type == 'string':
            return decoder.read_utf8()
        elif writers_schema.type == 'int':
            if logical_type == constants.DATE:
                return decoder.read_date_from_int()
            if logical_type == constants.TIME_MILLIS:
                return decoder.read_time_millis_from_int()
            return decoder.read_int()
        elif writers_schema.type == 'long':
            if logical_type == constants.TIME_MICROS:
                return decoder.read_time_micros_from_long()
            elif logical_type == constants.TIMESTAMP_MILLIS:
                return decoder.read_timestamp_millis_from_long()
            elif logical_type == constants.TIMESTAMP_MICROS:
                return decoder.read_timestamp_micros_from_long()
            else:
                return decoder.read_long()
        elif writers_schema.type == 'float':
            return decoder.read_float()
        elif writers_schema.type == 'double':
            return decoder.read_double()
        elif writers_schema.type == 'bytes':
            if logical_type == 'decimal':
                return decoder.read_decimal_from_bytes(
                    writers_schema.get_prop('precision'),
                    writers_schema.get_prop('scale')
                )
            else:
                return decoder.read_bytes()
        elif writers_schema.type == 'fixed':
            if logical_type == 'decimal':
                return decoder.read_decimal_from_fixed(
                    writers_schema.get_prop('precision'),
                    writers_schema.get_prop('scale'),
                    writers_schema.size
                )
            return self.read_fixed(writers_schema, readers_schema, decoder)
        elif writers_schema.type == 'enum':
            return self.read_enum(writers_schema, readers_schema, decoder)
        elif writers_schema.type == 'array':
            return self.read_array(writers_schema, readers_schema, decoder)
        elif writers_schema.type == 'map':
            return self.read_map(writers_schema, readers_schema, decoder)
        elif writers_schema.type in ['record', 'error', 'request']:
            return self.read_record(writers_schema, readers_schema, decoder)
        else:
            fail_msg = "Cannot read unknown schema type: %s" % writers_schema.type
            raise schema.AvroException(fail_msg)

    def skip_data(self, writers_schema, decoder):
        if writers_schema.type == 'null':
            return decoder.skip_null()
        elif writers_schema.type == 'boolean':
            return decoder.skip_boolean()
        elif writers_schema.type == 'string':
            return decoder.skip_utf8()
        elif writers_schema.type == 'int':
            return decoder.skip_int()
        elif writers_schema.type == 'long':
            return decoder.skip_long()
        elif writers_schema.type == 'float':
            return decoder.skip_float()
        elif writers_schema.type == 'double':
            return decoder.skip_double()
        elif writers_schema.type == 'bytes':
            return decoder.skip_bytes()
        elif writers_schema.type == 'fixed':
            return self.skip_fixed(writers_schema, decoder)
        elif writers_schema.type == 'enum':
            return self.skip_enum(writers_schema, decoder)
        elif writers_schema.type == 'array':
            return self.skip_array(writers_schema, decoder)
        elif writers_schema.type == 'map':
            return self.skip_map(writers_schema, decoder)
        elif writers_schema.type in ['union', 'error_union']:
            return self.skip_union(writers_schema, decoder)
        elif writers_schema.type in ['record', 'error', 'request']:
            return self.skip_record(writers_schema, decoder)
        else:
            fail_msg = "Unknown schema type: %s" % writers_schema.type
            raise schema.AvroException(fail_msg)

    def read_fixed(self, writers_schema, readers_schema, decoder):
        """
        Fixed instances are encoded using the number of bytes declared
        in the schema.
        """
        return decoder.read(writers_schema.size)

    def skip_fixed(self, writers_schema, decoder):
        return decoder.skip(writers_schema.size)

    def read_enum(self, writers_schema, readers_schema, decoder):
        """
        An enum is encoded by a int, representing the zero-based position
        of the symbol in the schema.
        """
        # read data
        index_of_symbol = decoder.read_int()
        if index_of_symbol >= len(writers_schema.symbols):
            fail_msg = "Can't access enum index %d for enum with %d symbols"\
                       % (index_of_symbol, len(writers_schema.symbols))
            raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)
        read_symbol = writers_schema.symbols[index_of_symbol]

        # schema resolution
        if read_symbol not in readers_schema.symbols:
            fail_msg = "Symbol %s not present in Reader's Schema" % read_symbol
            raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)

        return read_symbol

    def skip_enum(self, writers_schema, decoder):
        return decoder.skip_int()

    def read_array(self, writers_schema, readers_schema, decoder):
        """
        Arrays are encoded as a series of blocks.

        Each block consists of a long count value,
        followed by that many array items.
        A block with count zero indicates the end of the array.
        Each item is encoded per the array's item schema.

        If a block's count is negative,
        then the count is followed immediately by a long block size,
        indicating the number of bytes in the block.
        The actual count in this case
        is the absolute value of the count written.
        """
        read_items = []
        block_count = decoder.read_long()
        while block_count != 0:
            if block_count < 0:
                block_count = -block_count
                block_size = decoder.read_long()
            for i in range(block_count):
                read_items.append(self.read_data(writers_schema.items,
                                                 readers_schema.items, decoder))
            block_count = decoder.read_long()
        return read_items

    def skip_array(self, writers_schema, decoder):
        block_count = decoder.read_long()
        while block_count != 0:
            if block_count < 0:
                block_size = decoder.read_long()
                decoder.skip(block_size)
            else:
                for i in range(block_count):
                    self.skip_data(writers_schema.items, decoder)
            block_count = decoder.read_long()

    def read_map(self, writers_schema, readers_schema, decoder):
        """
        Maps are encoded as a series of blocks.

        Each block consists of a long count value,
        followed by that many key/value pairs.
        A block with count zero indicates the end of the map.
        Each item is encoded per the map's value schema.

        If a block's count is negative,
        then the count is followed immediately by a long block size,
        indicating the number of bytes in the block.
        The actual count in this case
        is the absolute value of the count written.
        """
        read_items = {}
        block_count = decoder.read_long()
        while block_count != 0:
            if block_count < 0:
                block_count = -block_count
                block_size = decoder.read_long()
            for i in range(block_count):
                key = decoder.read_utf8()
                read_items[key] = self.read_data(writers_schema.values,
                                                 readers_schema.values, decoder)
            block_count = decoder.read_long()
        return read_items

    def skip_map(self, writers_schema, decoder):
        block_count = decoder.read_long()
        while block_count != 0:
            if block_count < 0:
                block_size = decoder.read_long()
                decoder.skip(block_size)
            else:
                for i in range(block_count):
                    decoder.skip_utf8()
                    self.skip_data(writers_schema.values, decoder)
            block_count = decoder.read_long()

    def read_union(self, writers_schema, readers_schema, decoder):
        """
        A union is encoded by first writing a long value indicating
        the zero-based position within the union of the schema of its value.
        The value is then encoded per the indicated schema within the union.
        """
        # schema resolution
        index_of_schema = int(decoder.read_long())
        if index_of_schema >= len(writers_schema.schemas):
            fail_msg = "Can't access branch index %d for union with %d branches"\
                       % (index_of_schema, len(writers_schema.schemas))
            raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)
        selected_writers_schema = writers_schema.schemas[index_of_schema]

        # read data
        return self.read_data(selected_writers_schema, readers_schema, decoder)

    def skip_union(self, writers_schema, decoder):
        index_of_schema = int(decoder.read_long())
        if index_of_schema >= len(writers_schema.schemas):
            fail_msg = "Can't access branch index %d for union with %d branches"\
                       % (index_of_schema, len(writers_schema.schemas))
            raise SchemaResolutionException(fail_msg, writers_schema)
        return self.skip_data(writers_schema.schemas[index_of_schema], decoder)

    def read_record(self, writers_schema, readers_schema, decoder):
        """
        A record is encoded by encoding the values of its fields
        in the order that they are declared. In other words, a record
        is encoded as just the concatenation of the encodings of its fields.
        Field values are encoded per their schema.

        Schema Resolution:
         * the ordering of fields may be different: fields are matched by name.
         * schemas for fields with the same name in both records are resolved
           recursively.
         * if the writer's record contains a field with a name not present in the
           reader's record, the writer's value for that field is ignored.
         * if the reader's record schema has a field that contains a default value,
           and writer's schema does not have a field with the same name, then the
           reader should use the default value from its field.
         * if the reader's record schema has a field with no default value, and
           writer's schema does not have a field with the same name, then the
           field's value is unset.
        """
        # schema resolution
        readers_fields_dict = readers_schema.fields_dict
        read_record = {}
        for field in writers_schema.fields:
            readers_field = readers_fields_dict.get(field.name)
            if readers_field is not None:
                field_val = self.read_data(field.type, readers_field.type, decoder)
                read_record[field.name] = field_val
            else:
                self.skip_data(field.type, decoder)

        # fill in default values
        if len(readers_fields_dict) > len(read_record):
            writers_fields_dict = writers_schema.fields_dict
            for field_name, field in readers_fields_dict.items():
                if field_name not in writers_fields_dict:
                    if field.has_default:
                        field_val = self._read_default_value(field.type, field.default)
                        read_record[field.name] = field_val
                    else:
                        fail_msg = 'No default value for field %s' % field_name
                        raise SchemaResolutionException(fail_msg, writers_schema,
                                                        readers_schema)
        return read_record

    def skip_record(self, writers_schema, decoder):
        for field in writers_schema.fields:
            self.skip_data(field.type, decoder)

    def _read_default_value(self, field_schema, default_value):
        """
        Basically a JSON Decoder?
        """
        if field_schema.type == 'null':
            return None
        elif field_schema.type == 'boolean':
            return bool(default_value)
        elif field_schema.type == 'int':
            return int(default_value)
        elif field_schema.type == 'long':
            return long(default_value)
        elif field_schema.type in ['float', 'double']:
            return float(default_value)
        elif field_schema.type in ['enum', 'fixed', 'string', 'bytes']:
            return default_value
        elif field_schema.type == 'array':
            read_array = []
            for json_val in default_value:
                item_val = self._read_default_value(field_schema.items, json_val)
                read_array.append(item_val)
            return read_array
        elif field_schema.type == 'map':
            read_map = {}
            for key, json_val in default_value.items():
                map_val = self._read_default_value(field_schema.values, json_val)
                read_map[key] = map_val
            return read_map
        elif field_schema.type in ['union', 'error_union']:
            return self._read_default_value(field_schema.schemas[0], default_value)
        elif field_schema.type == 'record':
            read_record = {}
            for field in field_schema.fields:
                json_val = default_value.get(field.name)
                if json_val is None:
                    json_val = field.default
                field_val = self._read_default_value(field.type, json_val)
                read_record[field.name] = field_val
            return read_record
        else:
            fail_msg = 'Unknown type: %s' % field_schema.type
            raise schema.AvroException(fail_msg)


class DatumWriter(object):
    """DatumWriter for generic python objects."""

    def __init__(self, writers_schema=None):
        self._writers_schema = writers_schema

    # read/write properties
    def set_writers_schema(self, writers_schema):
        self._writers_schema = writers_schema
    writers_schema = property(lambda self: self._writers_schema,
                              set_writers_schema)

    def write(self, datum, encoder):
        if not validate(self.writers_schema, datum):
            raise AvroTypeException(self.writers_schema, datum)
        self.write_data(self.writers_schema, datum, encoder)

    def write_data(self, writers_schema, datum, encoder):
        # function dispatch to write datum
        logical_type = getattr(writers_schema, 'logical_type', None)
        if writers_schema.type == 'null':
            encoder.write_null(datum)
        elif writers_schema.type == 'boolean':
            encoder.write_boolean(datum)
        elif writers_schema.type == 'string':
            encoder.write_utf8(datum)
        elif writers_schema.type == 'int':
            if logical_type == constants.DATE:
                encoder.write_date_int(datum)
            elif logical_type == constants.TIME_MILLIS:
                encoder.write_time_millis_int(datum)
            else:
                encoder.write_int(datum)
        elif writers_schema.type == 'long':
            if logical_type == constants.TIME_MICROS:
                encoder.write_time_micros_long(datum)
            elif logical_type == constants.TIMESTAMP_MILLIS:
                encoder.write_timestamp_millis_long(datum)
            elif logical_type == constants.TIMESTAMP_MICROS:
                encoder.write_timestamp_micros_long(datum)
            else:
                encoder.write_long(datum)
        elif writers_schema.type == 'float':
            encoder.write_float(datum)
        elif writers_schema.type == 'double':
            encoder.write_double(datum)
        elif writers_schema.type == 'bytes':
            if logical_type == 'decimal':
                encoder.write_decimal_bytes(datum, writers_schema.get_prop('scale'))
            else:
                encoder.write_bytes(datum)
        elif writers_schema.type == 'fixed':
            if logical_type == 'decimal':
                encoder.write_decimal_fixed(
                    datum,
                    writers_schema.get_prop('scale'),
                    writers_schema.get_prop('size')
                )
            else:
                self.write_fixed(writers_schema, datum, encoder)
        elif writers_schema.type == 'enum':
            self.write_enum(writers_schema, datum, encoder)
        elif writers_schema.type == 'array':
            self.write_array(writers_schema, datum, encoder)
        elif writers_schema.type == 'map':
            self.write_map(writers_schema, datum, encoder)
        elif writers_schema.type in ['union', 'error_union']:
            self.write_union(writers_schema, datum, encoder)
        elif writers_schema.type in ['record', 'error', 'request']:
            self.write_record(writers_schema, datum, encoder)
        else:
            fail_msg = 'Unknown type: %s' % writers_schema.type
            raise schema.AvroException(fail_msg)

    def write_fixed(self, writers_schema, datum, encoder):
        """
        Fixed instances are encoded using the number of bytes declared
        in the schema.
        """
        encoder.write(datum)

    def write_enum(self, writers_schema, datum, encoder):
        """
        An enum is encoded by a int, representing the zero-based position
        of the symbol in the schema.
        """
        index_of_datum = writers_schema.symbols.index(datum)
        encoder.write_int(index_of_datum)

    def write_array(self, writers_schema, datum, encoder):
        """
        Arrays are encoded as a series of blocks.

        Each block consists of a long count value,
        followed by that many array items.
        A block with count zero indicates the end of the array.
        Each item is encoded per the array's item schema.

        If a block's count is negative,
        then the count is followed immediately by a long block size,
        indicating the number of bytes in the block.
        The actual count in this case
        is the absolute value of the count written.
        """
        if len(datum) > 0:
            encoder.write_long(len(datum))
            for item in datum:
                self.write_data(writers_schema.items, item, encoder)
        encoder.write_long(0)

    def write_map(self, writers_schema, datum, encoder):
        """
        Maps are encoded as a series of blocks.

        Each block consists of a long count value,
        followed by that many key/value pairs.
        A block with count zero indicates the end of the map.
        Each item is encoded per the map's value schema.

        If a block's count is negative,
        then the count is followed immediately by a long block size,
        indicating the number of bytes in the block.
        The actual count in this case
        is the absolute value of the count written.
        """
        if len(datum) > 0:
            encoder.write_long(len(datum))
            for key, val in datum.items():
                encoder.write_utf8(key)
                self.write_data(writers_schema.values, val, encoder)
        encoder.write_long(0)

    def write_union(self, writers_schema, datum, encoder):
        """
        A union is encoded by first writing a long value indicating
        the zero-based position within the union of the schema of its value.
        The value is then encoded per the indicated schema within the union.
        """
        # resolve union
        index_of_schema = -1
        for i, candidate_schema in enumerate(writers_schema.schemas):
            if validate(candidate_schema, datum):
                index_of_schema = i
        if index_of_schema < 0:
            raise AvroTypeException(writers_schema, datum)

        # write data
        encoder.write_long(index_of_schema)
        self.write_data(writers_schema.schemas[index_of_schema], datum, encoder)

    def write_record(self, writers_schema, datum, encoder):
        """
        A record is encoded by encoding the values of its fields
        in the order that they are declared. In other words, a record
        is encoded as just the concatenation of the encodings of its fields.
        Field values are encoded per their schema.
        """
        for field in writers_schema.fields:
            self.write_data(field.type, datum.get(field.name), encoder)