# 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
# 
# 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.
"""
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. 
"""
import struct
from avro import schema
import sys
from binascii import crc32

try:
	import json
except ImportError:
	import simplejson as json

#
# Constants
#

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?)
if sys.version_info >= (2, 5, 0):
  struct_class = struct.Struct
else:
  class SimpleStruct(object):
    def __init__(self, format):
      self.format = format
    def pack(self, *args):
      return struct.pack(self.format, *args)
    def unpack(self, *args):
      return struct.unpack(self.format, *args)
  struct_class = SimpleStruct

STRUCT_INT = struct_class('!I')     # big-endian unsigned int
STRUCT_LONG = struct_class('!Q')    # big-endian unsigned long long
STRUCT_FLOAT = struct_class('!f')   # big-endian float
STRUCT_DOUBLE = struct_class('!d')  # big-endian double
STRUCT_CRC32 = struct_class('>I')   # big-endian unsigned int

#
# 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 validate(expected_schema, datum):
  """Determine if a python datum is an instance of a schema."""
  schema_type = expected_schema.type
  if schema_type == 'null':
    return datum is None
  elif schema_type == 'boolean':
    return isinstance(datum, bool)
  elif schema_type == 'string':
    return isinstance(datum, basestring)
  elif schema_type == 'bytes':
    return isinstance(datum, str)
  elif schema_type == 'int':
    return ((isinstance(datum, int) or isinstance(datum, long)) 
            and INT_MIN_VALUE <= datum <= INT_MAX_VALUE)
  elif schema_type == 'long':
    return ((isinstance(datum, int) or isinstance(datum, long)) 
            and LONG_MIN_VALUE <= datum <= LONG_MAX_VALUE)
  elif schema_type in ['float', 'double']:
    return (isinstance(datum, int) or isinstance(datum, long)
            or isinstance(datum, float))
  elif schema_type == 'fixed':
    return isinstance(datum, str) and len(datum) == expected_schema.size
  elif schema_type == 'enum':
    return datum in expected_schema.symbols
  elif schema_type == 'array':
    return (isinstance(datum, list) and
      False not in [validate(expected_schema.items, d) for d in datum])
  elif schema_type == 'map':
    return (isinstance(datum, dict) and
      False not in [isinstance(k, basestring) for k in datum.keys()] and
      False not in
        [validate(expected_schema.values, v) for v in datum.values()])
  elif schema_type in ['union', 'error_union']:
    return True in [validate(s, datum) for s in expected_schema.schemas]
  elif schema_type in ['record', 'error', 'request']:
    return (isinstance(datum, dict) and
      False not in
        [validate(f.type, datum.get(f.name)) for f in expected_schema.fields])

#
# 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.
    """
    bits = (((ord(self.read(1)) & 0xffL)) |
      ((ord(self.read(1)) & 0xffL) <<  8) |
      ((ord(self.read(1)) & 0xffL) << 16) |
      ((ord(self.read(1)) & 0xffL) << 24))
    return STRUCT_FLOAT.unpack(STRUCT_INT.pack(bits))[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.
    """
    bits = (((ord(self.read(1)) & 0xffL)) |
      ((ord(self.read(1)) & 0xffL) <<  8) |
      ((ord(self.read(1)) & 0xffL) << 16) |
      ((ord(self.read(1)) & 0xffL) << 24) |
      ((ord(self.read(1)) & 0xffL) << 32) |
      ((ord(self.read(1)) & 0xffL) << 40) |
      ((ord(self.read(1)) & 0xffL) << 48) |
      ((ord(self.read(1)) & 0xffL) << 56))
    return STRUCT_DOUBLE.unpack(STRUCT_LONG.pack(bits))[0]

  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 check_crc32(self, bytes):
    checksum = STRUCT_CRC32.unpack(self.read(4))[0];
    if crc32(bytes) & 0xffffffff != checksum:
      raise schema.AvroException("Checksum failure")

  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 abritrary 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).
    """
    if datum:
      self.write(chr(1))
    else:
      self.write(chr(0))

  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(chr((datum & 0x7f) | 0x80))
      datum >>= 7
    self.write(chr(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.
    """
    bits = STRUCT_INT.unpack(STRUCT_FLOAT.pack(datum))[0]
    self.write(chr((bits) & 0xFF))
    self.write(chr((bits >> 8) & 0xFF))
    self.write(chr((bits >> 16) & 0xFF))
    self.write(chr((bits >> 24) & 0xFF))

  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.
    """
    bits = STRUCT_LONG.unpack(STRUCT_DOUBLE.pack(datum))[0]
    self.write(chr((bits) & 0xFF))
    self.write(chr((bits >> 8) & 0xFF))
    self.write(chr((bits >> 16) & 0xFF))
    self.write(chr((bits >> 24) & 0xFF))
    self.write(chr((bits >> 32) & 0xFF))
    self.write(chr((bits >> 40) & 0xFF))
    self.write(chr((bits >> 48) & 0xFF))
    self.write(chr((bits >> 56) & 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_crc32(self, bytes):
    """
    A 4-byte, big-endian CRC32 checksum
    """
    self.write(STRUCT_CRC32.pack(crc32(bytes) & 0xffffffff));

#
# DatumReader/Writer
#

class DatumReader(object):
  """Deserialize Avro-encoded data into a Python data structure."""
  @staticmethod
  def check_props(schema_one, schema_two, prop_list):
    for prop in prop_list:
      if getattr(schema_one, prop) != getattr(schema_two, prop):
        return False
    return True

  @staticmethod
  def match_schemas(writers_schema, readers_schema):
    w_type = writers_schema.type
    r_type = readers_schema.type
    if 'union' in [w_type, r_type] or 'error_union' in [w_type, r_type]:
      return True
    elif (w_type in schema.PRIMITIVE_TYPES and r_type in schema.PRIMITIVE_TYPES
          and w_type == r_type):
      return True
    elif (w_type == r_type == 'record' and
          DatumReader.check_props(writers_schema, readers_schema, 
                                  ['fullname'])):
      return True
    elif (w_type == r_type == 'error' and
          DatumReader.check_props(writers_schema, readers_schema, 
                                  ['fullname'])):
      return True
    elif (w_type == r_type == 'request'):
      return True
    elif (w_type == r_type == 'fixed' and 
          DatumReader.check_props(writers_schema, readers_schema, 
                                  ['fullname', 'size'])):
      return True
    elif (w_type == r_type == 'enum' and 
          DatumReader.check_props(writers_schema, readers_schema, 
                                  ['fullname'])):
      return True
    elif (w_type == r_type == 'map' and 
          DatumReader.check_props(writers_schema.values,
                                  readers_schema.values, ['type'])):
      return True
    elif (w_type == r_type == 'array' and 
          DatumReader.check_props(writers_schema.items,
                                  readers_schema.items, ['type'])):
      return True
    
    # Handle schema promotion
    if w_type == 'int' and r_type in ['long', 'float', 'double']:
      return True
    elif w_type == 'long' and r_type in ['float', 'double']:
      return True
    elif w_type == 'float' and r_type == 'double':
      return True
    return False

  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 DatumReader.match_schemas(writers_schema, readers_schema):
      fail_msg = 'Schemas do not match.'
      raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)

    # schema resolution: reader's schema is a union, writer's schema is not
    if (writers_schema.type not in ['union', 'error_union']
        and readers_schema.type in ['union', 'error_union']):
      for s in readers_schema.schemas:
        if DatumReader.match_schemas(writers_schema, s):
          return self.read_data(writers_schema, s, decoder)
      fail_msg = 'Schemas do not match.'
      raise SchemaResolutionException(fail_msg, writers_schema, readers_schema)

    # function dispatch for reading data based on type of writer's 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':
      return decoder.read_int()
    elif writers_schema.type == 'long':
      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':
      return decoder.read_bytes()
    elif writers_schema.type == 'fixed':
      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 ['union', 'error_union']:
      return self.read_union(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 not writers_fields_dict.has_key(field_name):
          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):
    # validate datum
    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
    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':
      encoder.write_int(datum)
    elif writers_schema.type == 'long':
      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':
      encoder.write_bytes(datum)
    elif writers_schema.type == 'fixed':
      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)