#!/usr/bin/env python3

##
# 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 str.
  * Schema bytes are implemented as bytes.
  * Schema ints are implemented as int.
  * Schema longs are implemented as int.
  * Schema floats are implemented as float.
  * Schema doubles are implemented as float.
  * Schema booleans are implemented as bool.

Validation:

The validation of schema is performed using breadth-first graph
traversal. This allows validation exceptions to pinpoint the exact node
within a complex schema that is problematic, simplifying debugging
considerably. Because it is a traversal, it will also be less
resource-intensive, particularly when validating schema with deep
structures.

Components
==========

Nodes
-----
Avro schemas contain many different schema types. Data about the schema
types is used to validate the data in the corresponding part of a Python
body (the object to be serialized). A node combines a given schema type
with the corresponding Python data, as well as an optional "name" to
identify the specific node. Names are generally the name of a schema
(for named schema) or the name of a field (for child nodes of schema
with named children like maps and records), or None, for schema who's
children are not named (like Arrays).

Iterators
---------
Iterators are generator functions that take a node and return a
generator which will yield a node for each child datum in the data for
the current node. If a node is of a type which has no children, then the
default iterator will immediately exit.

Validators
----------
Validators are used to determine if the datum for a given node is valid
according to the given schema type. Validator functions take a node as
an argument and return a node if the node datum passes validation. If it
does not, the validator must return None.

In most cases, the node returned is identical to the node provided (is
in fact the same object). However, in the case of Union schema, the
returned "valid" node will hold the schema that is represented by the
datum contained. This allows iteration over the child nodes
in that datum, if there are any.
"""

import collections
import datetime
import decimal
import struct
import warnings
from typing import IO, Generator, Iterable, List, Mapping, Optional, Sequence, Union

import avro.constants
import avro.errors
import avro.schema
import avro.timezones

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

ValidationNode = collections.namedtuple("ValidationNode", ["schema", "datum", "name"])
ValidationNodeGeneratorType = Generator[ValidationNode, None, None]
JsonScalarFieldType = Union[None, bool, str, int, float]


def validate(expected_schema: avro.schema.Schema, datum: object, raise_on_error: bool = False) -> bool:
    """Return True if the provided datum is valid for the expected schema

    If raise_on_error is passed and True, then raise a validation error
    with specific information about the error encountered in validation.

    :param expected_schema: An avro schema type object representing the schema against
                            which the datum will be validated.
    :param datum: The datum to be validated, A python dictionary or some supported type
    :param raise_on_error: True if a AvroTypeException should be raised immediately when a
                           validation problem is encountered.
    :raises: AvroTypeException if datum is invalid and raise_on_error is True
    :returns: True if datum is valid for expected_schema, False if not.
    """
    # use a FIFO queue to process schema nodes breadth first.
    nodes = collections.deque([ValidationNode(expected_schema, datum, getattr(expected_schema, "name", None))])

    while nodes:
        current_node = nodes.popleft()

        # _validate_node returns the node for iteration if it is valid. Or it returns None
        validated_schema = current_node.schema.validate(current_node.datum)
        valid_node = ValidationNode(validated_schema, current_node.datum, current_node.name) if validated_schema else None

        if valid_node is None:
            if raise_on_error:
                raise avro.errors.AvroTypeException(current_node.schema, current_node.name, current_node.datum)
            return False  # preserve the prior validation behavior of returning false when there are problems.
        # if there are children of this node to append, do so.
        for child_node in _iterate_node(valid_node):
            nodes.append(child_node)

    return True


def _iterate_node(node: ValidationNode) -> ValidationNodeGeneratorType:
    for item in _ITERATORS.get(node.schema.type, _default_iterator)(node):
        yield ValidationNode(*item)


#############
# Iteration #
#############


def _default_iterator(_) -> ValidationNodeGeneratorType:
    """Immediately raise StopIteration.

    This exists to prevent problems with iteration over unsupported container types.
    """
    yield from ()


def _record_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
    """Yield each child node of the provided record node."""
    schema, datum, _ = node
    return (ValidationNode(field.type, datum.get(field.name), field.name) for field in schema.fields)


def _array_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
    """Yield each child node of the provided array node."""
    schema, datum, name = node
    return (ValidationNode(schema.items, item, name) for item in datum)


def _map_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
    """Yield each child node of the provided map node."""
    schema, datum, _ = node
    child_schema = schema.values
    return (ValidationNode(child_schema, child_datum, child_name) for child_name, child_datum in datum.items())


_ITERATORS = {
    "record": _record_iterator,
    "array": _array_iterator,
    "map": _map_iterator,
}
_ITERATORS["error"] = _ITERATORS["request"] = _ITERATORS["record"]


#
# Decoder/Encoder
#


class BinaryDecoder:
    """Read leaf values."""

    _reader: IO[bytes]

    def __init__(self, reader: IO[bytes]) -> None:
        """
        reader is a Python object on which we can call read, seek, and tell.
        """
        self._reader = reader

    @property
    def reader(self) -> IO[bytes]:
        return self._reader

    def read(self, n: int) -> bytes:
        """
        Read n bytes.
        """
        if n < 0:
            raise avro.errors.InvalidAvroBinaryEncoding(f"Requested {n} bytes to read, expected positive integer.")
        read_bytes = self.reader.read(n)
        if len(read_bytes) != n:
            raise avro.errors.InvalidAvroBinaryEncoding(f"Read {len(read_bytes)} bytes, expected {n} bytes")
        return read_bytes

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

    def read_boolean(self) -> bool:
        """
        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:
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        return self.read_long()

    def read_long(self) -> int:
        """
        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) -> float:
        """
        A float is written as 4 bytes.
        The float is converted into a 32-bit integer using a method equivalent to
        Java's floatToRawIntBits and then encoded in little-endian format.
        """
        return float(STRUCT_FLOAT.unpack(self.read(4))[0])

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

    def read_decimal_from_bytes(self, precision: int, scale: int) -> decimal.Decimal:
        """
        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: int, scale: int, size: int) -> decimal.Decimal:
        """
        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 = decimal.getcontext().prec
        try:
            decimal.getcontext().prec = precision
            scaled_datum = decimal.Decimal(unscaled_datum).scaleb(-scale)
        finally:
            decimal.getcontext().prec = original_prec
        return scaled_datum

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

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

    def read_date_from_int(self) -> datetime.date:
        """
        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: int, scale_to_micro: int) -> datetime.time:
        value = value * scale_to_micro
        value, microseconds = divmod(value, 1000000)
        value, seconds = divmod(value, 60)
        value, minutes = divmod(value, 60)
        hours = value

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

    def read_time_millis_from_int(self) -> datetime.time:
        """
        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) -> datetime.time:
        """
        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) -> datetime.datetime:
        """
        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=avro.timezones.utc)
        return unix_epoch_datetime + timedelta

    def read_timestamp_micros_from_long(self) -> datetime.datetime:
        """
        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=avro.timezones.utc)
        return unix_epoch_datetime + timedelta

    def skip_null(self) -> None:
        pass

    def skip_boolean(self) -> None:
        self.skip(1)

    def skip_int(self) -> None:
        self.skip_long()

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

    def skip_float(self) -> None:
        self.skip(4)

    def skip_double(self) -> None:
        self.skip(8)

    def skip_bytes(self) -> None:
        self.skip(self.read_long())

    def skip_utf8(self) -> None:
        self.skip_bytes()

    def skip(self, n: int) -> None:
        self.reader.seek(self.reader.tell() + n)


class BinaryEncoder:
    """Write leaf values."""

    _writer: IO[bytes]

    def __init__(self, writer: IO[bytes]) -> None:
        """
        writer is a Python object on which we can call write.
        """
        self._writer = writer

    @property
    def writer(self) -> IO[bytes]:
        return self._writer

    def write(self, datum: bytes) -> None:
        """Write an arbitrary datum."""
        self.writer.write(datum)

    def write_null(self, datum: None) -> None:
        """
        null is written as zero bytes
        """

    def write_boolean(self, datum: bool) -> None:
        """
        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) -> None:
        """
        int and long values are written using variable-length, zig-zag coding.
        """
        self.write_long(datum)

    def write_long(self, datum: int) -> None:
        """
        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: float) -> None:
        """
        A float is written as 4 bytes.
        The float is converted into a 32-bit integer using a method equivalent to
        Java's floatToRawIntBits and then encoded in little-endian format.
        """
        self.write(STRUCT_FLOAT.pack(datum))

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

    def write_decimal_bytes(self, datum: decimal.Decimal, scale: int) -> None:
        """
        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 (-1 * int(exp)) > scale:
            raise avro.errors.AvroOutOfScaleException(scale, datum, exp)

        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: decimal.Decimal, scale: int, size: int) -> None:
        """
        Decimal in fixed are encoded as size of fixed bytes.
        """
        sign, digits, exp = datum.as_tuple()
        if (-1 * int(exp)) > scale:
            raise avro.errors.AvroOutOfScaleException(scale, datum, exp)

        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) -> None:
        """
        Bytes are encoded as a long followed by that many bytes of data.
        """
        self.write_long(len(datum))
        self.write(struct.pack(f"{len(datum)}s", datum))

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

    def write_date_int(self, datum: datetime.date) -> None:
        """
        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: datetime.time) -> None:
        """
        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: datetime.time) -> None:
        """
        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_: datetime.timedelta) -> int:
        return timedelta_.microseconds + (timedelta_.seconds + timedelta_.days * 24 * 3600) * 10**6

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

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


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

    _writers_schema: Optional[avro.schema.Schema]
    _readers_schema: Optional[avro.schema.Schema]

    def __init__(self, writers_schema: Optional[avro.schema.Schema] = None, readers_schema: Optional[avro.schema.Schema] = None) -> 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

    @property
    def writers_schema(self) -> Optional[avro.schema.Schema]:
        return self._writers_schema

    @writers_schema.setter
    def writers_schema(self, writers_schema: avro.schema.Schema) -> None:
        self._writers_schema = writers_schema

    @property
    def readers_schema(self) -> Optional[avro.schema.Schema]:
        return self._readers_schema

    @readers_schema.setter
    def readers_schema(self, readers_schema: avro.schema.Schema) -> None:
        self._readers_schema = readers_schema

    def read(self, decoder: "BinaryDecoder") -> object:
        if self.writers_schema is None:
            raise avro.errors.IONotReadyException("Cannot read without a writer's schema.")
        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: avro.schema.Schema, readers_schema: avro.schema.Schema, decoder: "BinaryDecoder") -> object:
        # schema matching
        if not readers_schema.match(writers_schema):
            raise avro.errors.SchemaResolutionException("Schemas do not match.", 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 isinstance(writers_schema, avro.schema.UnionSchema) and isinstance(readers_schema, avro.schema.UnionSchema):
            return self.read_union(writers_schema, readers_schema, decoder)

        if isinstance(readers_schema, avro.schema.UnionSchema):
            # 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.
            raise avro.errors.SchemaResolutionException("Schemas do not match.", writers_schema, readers_schema)

        if writers_schema.type == "null":
            return None
        if writers_schema.type == "boolean":
            return decoder.read_boolean()
        if writers_schema.type == "string":
            return decoder.read_utf8()
        if writers_schema.type == "int":
            if logical_type == avro.constants.DATE:
                return decoder.read_date_from_int()
            if logical_type == avro.constants.TIME_MILLIS:
                return decoder.read_time_millis_from_int()
            return decoder.read_int()
        if writers_schema.type == "long":
            if logical_type == avro.constants.TIME_MICROS:
                return decoder.read_time_micros_from_long()
            if logical_type == avro.constants.TIMESTAMP_MILLIS:
                return decoder.read_timestamp_millis_from_long()
            if logical_type == avro.constants.TIMESTAMP_MICROS:
                return decoder.read_timestamp_micros_from_long()
            return decoder.read_long()
        if writers_schema.type == "float":
            return decoder.read_float()
        if writers_schema.type == "double":
            return decoder.read_double()
        if writers_schema.type == "bytes":
            if logical_type == "decimal":
                precision = writers_schema.get_prop("precision")
                if not (isinstance(precision, int) and precision > 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal precision {precision}. Must be a positive integer."))
                    return decoder.read_bytes()
                scale = writers_schema.get_prop("scale")
                if not (isinstance(scale, int) and scale >= 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal scale {scale}. Must be a non-negative integer."))
                    return decoder.read_bytes()
                return decoder.read_decimal_from_bytes(precision, scale)
            return decoder.read_bytes()
        if isinstance(writers_schema, avro.schema.FixedSchema) and isinstance(readers_schema, avro.schema.FixedSchema):
            if logical_type == "decimal":
                precision = writers_schema.get_prop("precision")
                if not (isinstance(precision, int) and precision > 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal precision {precision}. Must be a positive integer."))
                    return self.read_fixed(writers_schema, readers_schema, decoder)
                scale = writers_schema.get_prop("scale")
                if not (isinstance(scale, int) and scale >= 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal scale {scale}. Must be a non-negative integer."))
                    return self.read_fixed(writers_schema, readers_schema, decoder)
                return decoder.read_decimal_from_fixed(precision, scale, writers_schema.size)
            return self.read_fixed(writers_schema, readers_schema, decoder)
        if isinstance(writers_schema, avro.schema.EnumSchema) and isinstance(readers_schema, avro.schema.EnumSchema):
            return self.read_enum(writers_schema, readers_schema, decoder)
        if isinstance(writers_schema, avro.schema.ArraySchema) and isinstance(readers_schema, avro.schema.ArraySchema):
            return self.read_array(writers_schema, readers_schema, decoder)
        if isinstance(writers_schema, avro.schema.MapSchema) and isinstance(readers_schema, avro.schema.MapSchema):
            return self.read_map(writers_schema, readers_schema, decoder)
        if isinstance(writers_schema, avro.schema.RecordSchema) and isinstance(readers_schema, avro.schema.RecordSchema):
            # .type in ["record", "error", "request"]:
            return self.read_record(writers_schema, readers_schema, decoder)
        raise avro.errors.AvroException(f"Cannot read unknown schema type: {writers_schema.type}")

    def skip_data(self, writers_schema: avro.schema.Schema, decoder: BinaryDecoder) -> None:
        if writers_schema.type == "null":
            return decoder.skip_null()
        if writers_schema.type == "boolean":
            return decoder.skip_boolean()
        if writers_schema.type == "string":
            return decoder.skip_utf8()
        if writers_schema.type == "int":
            return decoder.skip_int()
        if writers_schema.type == "long":
            return decoder.skip_long()
        if writers_schema.type == "float":
            return decoder.skip_float()
        if writers_schema.type == "double":
            return decoder.skip_double()
        if writers_schema.type == "bytes":
            return decoder.skip_bytes()
        if isinstance(writers_schema, avro.schema.FixedSchema):
            return self.skip_fixed(writers_schema, decoder)
        if isinstance(writers_schema, avro.schema.EnumSchema):
            return self.skip_enum(writers_schema, decoder)
        if isinstance(writers_schema, avro.schema.ArraySchema):
            return self.skip_array(writers_schema, decoder)
        if isinstance(writers_schema, avro.schema.MapSchema):
            return self.skip_map(writers_schema, decoder)
        if isinstance(writers_schema, avro.schema.UnionSchema):
            return self.skip_union(writers_schema, decoder)
        if isinstance(writers_schema, avro.schema.RecordSchema):
            return self.skip_record(writers_schema, decoder)
        raise avro.errors.AvroException(f"Unknown schema type: {writers_schema.type}")

    def read_fixed(self, writers_schema: avro.schema.FixedSchema, readers_schema: avro.schema.Schema, decoder: BinaryDecoder) -> bytes:
        """
        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: avro.schema.FixedSchema, decoder: BinaryDecoder) -> None:
        return decoder.skip(writers_schema.size)

    def read_enum(self, writers_schema: avro.schema.EnumSchema, readers_schema: avro.schema.EnumSchema, decoder: BinaryDecoder) -> str:
        """
        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):
            raise avro.errors.SchemaResolutionException(
                f"Can't access enum index {index_of_symbol} for enum with {len(writers_schema.symbols)} symbols", writers_schema, readers_schema
            )
        read_symbol = writers_schema.symbols[index_of_symbol]

        # schema resolution
        if read_symbol not in readers_schema.symbols:
            raise avro.errors.SchemaResolutionException(f"Symbol {read_symbol} not present in Reader's Schema", writers_schema, readers_schema)

        return read_symbol

    def skip_enum(self, writers_schema: avro.schema.EnumSchema, decoder: BinaryDecoder) -> None:
        return decoder.skip_int()

    def read_array(self, writers_schema: avro.schema.ArraySchema, readers_schema: avro.schema.ArraySchema, decoder: BinaryDecoder) -> List[object]:
        """
        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
                decoder.skip_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: avro.schema.ArraySchema, decoder: BinaryDecoder) -> None:
        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: avro.schema.MapSchema, readers_schema: avro.schema.MapSchema, decoder: BinaryDecoder) -> Mapping[str, object]:
        """
        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
                decoder.skip_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: avro.schema.MapSchema, decoder: BinaryDecoder) -> None:
        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: avro.schema.UnionSchema, readers_schema: avro.schema.UnionSchema, decoder: BinaryDecoder) -> object:
        """
        A union is encoded by first writing an int 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):
            raise avro.errors.SchemaResolutionException(
                f"Can't access branch index {index_of_schema} for union with {len(writers_schema.schemas)} branches", 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: avro.schema.UnionSchema, decoder: BinaryDecoder) -> None:
        index_of_schema = int(decoder.read_long())
        if index_of_schema >= len(writers_schema.schemas):
            raise avro.errors.SchemaResolutionException(
                f"Can't access branch index {index_of_schema} for union with {len(writers_schema.schemas)} branches", writers_schema
            )
        return self.skip_data(writers_schema.schemas[index_of_schema], decoder)

    def read_record(
        self, writers_schema: avro.schema.RecordSchema, readers_schema: avro.schema.RecordSchema, decoder: BinaryDecoder
    ) -> Mapping[str, object]:
        """
        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 not field.has_default:
                        raise avro.errors.SchemaResolutionException(f"No default value for field {field_name}", writers_schema, readers_schema)
                    field_val = self._read_default_value(field.type, field.default)
                    read_record[field.name] = field_val
        return read_record

    def skip_record(self, writers_schema: avro.schema.RecordSchema, decoder: BinaryDecoder) -> None:
        for field in writers_schema.fields:
            self.skip_data(field.type, decoder)

    def _read_default_value(self, field_schema: avro.schema.Schema, default_value: object) -> object:
        """
        Basically a JSON Decoder?
        """
        if field_schema.type == "null":
            if default_value is None:
                return None
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if field_schema.type == "boolean":
            return bool(default_value)
        if field_schema.type in ("int", "long"):
            if isinstance(default_value, int):
                return default_value
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if field_schema.type in ("float", "double"):
            if isinstance(default_value, float):
                return default_value
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if field_schema.type in ("bytes", "fixed"):
            if isinstance(default_value, bytes):
                return default_value
            if isinstance(default_value, str):
                return default_value.encode()
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if field_schema.type in ("enum", "string"):
            if isinstance(default_value, str):
                return default_value
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if isinstance(field_schema, avro.schema.ArraySchema):
            if isinstance(default_value, Iterable):
                return [self._read_default_value(field_schema.items, json_val) for json_val in default_value]
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if isinstance(field_schema, avro.schema.MapSchema):
            if isinstance(default_value, Mapping):
                return {key: self._read_default_value(field_schema.values, json_val) for key, json_val in default_value.items()}
            raise avro.errors.InvalidDefaultException(field_schema, default_value)
        if isinstance(field_schema, avro.schema.UnionSchema):
            return self._read_default_value(field_schema.schemas[0], default_value)
        if isinstance(field_schema, avro.schema.RecordSchema):
            if not isinstance(default_value, Mapping):
                raise avro.errors.InvalidDefaultException(field_schema, default_value)
            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
        raise avro.errors.AvroException(f"Unknown type: {field_schema.type}")


class DatumWriter:
    """DatumWriter for generic python objects."""

    _writers_schema: Optional[avro.schema.Schema]

    def __init__(self, writers_schema: Optional[avro.schema.Schema] = None) -> None:
        self._writers_schema = writers_schema

    @property
    def writers_schema(self) -> Optional[avro.schema.Schema]:
        return self._writers_schema

    @writers_schema.setter
    def writers_schema(self, writers_schema: avro.schema.Schema) -> None:
        self._writers_schema = writers_schema

    def write(self, datum: object, encoder: BinaryEncoder) -> None:
        if self.writers_schema is None:
            raise avro.errors.IONotReadyException("Cannot write without a writer's schema.")
        validate(self.writers_schema, datum, raise_on_error=True)
        self.write_data(self.writers_schema, datum, encoder)

    def write_data(self, writers_schema: avro.schema.Schema, datum: object, encoder: BinaryEncoder) -> None:
        # function dispatch to write datum
        logical_type = getattr(writers_schema, "logical_type", None)
        if writers_schema.type == "null":
            if datum is None:
                return encoder.write_null(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "boolean":
            if isinstance(datum, bool):
                return encoder.write_boolean(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "string":
            if isinstance(datum, str):
                return encoder.write_utf8(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "int":
            if logical_type == avro.constants.DATE:
                if isinstance(datum, datetime.date):
                    return encoder.write_date_int(datum)
                warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a date type"))
            elif logical_type == avro.constants.TIME_MILLIS:
                if isinstance(datum, datetime.time):
                    return encoder.write_time_millis_int(datum)
                warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a time type"))
            if isinstance(datum, int):
                return encoder.write_int(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "long":
            if logical_type == avro.constants.TIME_MICROS:
                if isinstance(datum, datetime.time):
                    return encoder.write_time_micros_long(datum)
                warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a time type"))
            elif logical_type == avro.constants.TIMESTAMP_MILLIS:
                if isinstance(datum, datetime.datetime):
                    return encoder.write_timestamp_millis_long(datum)
                warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a datetime type"))
            elif logical_type == avro.constants.TIMESTAMP_MICROS:
                if isinstance(datum, datetime.datetime):
                    return encoder.write_timestamp_micros_long(datum)
                warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a datetime type"))
            if isinstance(datum, int):
                return encoder.write_long(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "float":
            if isinstance(datum, (int, float)):
                return encoder.write_float(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "double":
            if isinstance(datum, (int, float)):
                return encoder.write_double(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if writers_schema.type == "bytes":
            if logical_type == "decimal":
                scale = writers_schema.get_prop("scale")
                if not (isinstance(scale, int) and scale >= 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal scale {scale}. Must be a non-negative integer."))
                elif not isinstance(datum, decimal.Decimal):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a decimal type"))
                else:
                    return encoder.write_decimal_bytes(datum, scale)
            if isinstance(datum, bytes):
                return encoder.write_bytes(datum)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if isinstance(writers_schema, avro.schema.FixedSchema):
            if logical_type == "decimal":
                scale = writers_schema.get_prop("scale")
                size = writers_schema.size
                if not (isinstance(scale, int) and scale >= 0):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"Invalid decimal scale {scale}. Must be a non-negative integer."))
                elif not isinstance(datum, decimal.Decimal):
                    warnings.warn(avro.errors.IgnoredLogicalType(f"{datum} is not a decimal type"))
                else:
                    return encoder.write_decimal_fixed(datum, scale, size)
            if isinstance(datum, bytes):
                return self.write_fixed(writers_schema, datum, encoder)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if isinstance(writers_schema, avro.schema.EnumSchema):
            if isinstance(datum, str):
                return self.write_enum(writers_schema, datum, encoder)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if isinstance(writers_schema, avro.schema.ArraySchema):
            if isinstance(datum, Sequence):
                return self.write_array(writers_schema, datum, encoder)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if isinstance(writers_schema, avro.schema.MapSchema):
            if isinstance(datum, Mapping):
                return self.write_map(writers_schema, datum, encoder)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        if isinstance(writers_schema, avro.schema.UnionSchema):
            return self.write_union(writers_schema, datum, encoder)
        if isinstance(writers_schema, avro.schema.RecordSchema):
            if isinstance(datum, Mapping):
                return self.write_record(writers_schema, datum, encoder)
            raise avro.errors.AvroTypeException(writers_schema, datum)
        raise avro.errors.AvroException(f"Unknown type: {writers_schema.type}")

    def write_fixed(self, writers_schema: avro.schema.FixedSchema, datum: bytes, encoder: BinaryEncoder) -> None:
        """
        Fixed instances are encoded using the number of bytes declared
        in the schema.
        """
        return encoder.write(datum)

    def write_enum(self, writers_schema: avro.schema.EnumSchema, datum: str, encoder: BinaryEncoder) -> None:
        """
        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)
        return encoder.write_int(index_of_datum)

    def write_array(self, writers_schema: avro.schema.ArraySchema, datum: Sequence[object], encoder: BinaryEncoder) -> None:
        """
        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)
        return encoder.write_long(0)

    def write_map(self, writers_schema: avro.schema.MapSchema, datum: Mapping[str, object], encoder: BinaryEncoder) -> None:
        """
        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)
        return encoder.write_long(0)

    def write_union(self, writers_schema: avro.schema.UnionSchema, datum: object, encoder: BinaryEncoder) -> None:
        """
        A union is encoded by first writing an int 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 avro.errors.AvroTypeException(writers_schema, datum)

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

    def write_record(self, writers_schema: avro.schema.RecordSchema, datum: Mapping[str, object], encoder: BinaryEncoder) -> None:
        """
        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)