class BinaryDecoder:
    """Decoder for the avro binary format.

    NOTE: All attributes and methods on this class should be considered
    private.

    Parameters
    ----------
    fo: file-like
        Input stream

    """

    def __init__(self, fo):
        self.fo = fo

    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).
        """

        # technically 0x01 == true and 0x00 == false, but many languages will
        # cast anything other than 0 to True and only 0 to False
        return unpack("B", self.fo.read(1))[0] != 0

    def read_long(self):
        """int and long values are written using variable-length, zig-zag
        coding."""
        c = self.fo.read(1)

        # We do EOF checking only here, since most reader start here
        if not c:
            raise StopIteration

        b = ord(c)
        n = b & 0x7F
        shift = 7

        while (b & 0x80) != 0:
            b = ord(self.fo.read(1))
            n |= (b & 0x7F) << shift
            shift += 7

        return (n >> 1) ^ -(n & 1)

    read_int = read_long

    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 unpack("<f", self.fo.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 unpack("<d", self.fo.read(8))[0]

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

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

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

    def read_enum(self):
        """An enum is encoded by a int, representing the zero-based position of the
        symbol in the schema.
        """
        return self.read_long()

    def read_array_start(self):
        """Arrays are encoded as a series of blocks."""
        self._block_count = self.read_long()

    def read_array_end(self):
        pass

    def _iter_array_or_map(self):
        """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.
        """
        while self._block_count != 0:
            if self._block_count < 0:
                self._block_count = -self._block_count
                # Read block size, unused
                self.read_long()

            for i in range(self._block_count):
                yield
            self._block_count = self.read_long()

    iter_array = _iter_array_or_map
    iter_map = _iter_array_or_map

    def read_map_start(self):
        """Maps are encoded as a series of blocks."""
        self._block_count = self.read_long()

    def read_map_end(self):
        pass

    def read_index(self):
        """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.
        """
        return self.read_long() class AvroJSONDecoder:
    """Decoder for the avro JSON format.

    NOTE: All attributes and methods on this class should be considered
    private.

    Parameters
    ----------
    fo: file-like
        Input stream

    """

    def __init__(self, fo):
        self._fo = fo
        self._stack = []
        self._json_data = [json.loads(line.strip()) for line in fo]
        if self._json_data:
            self._current = self._json_data.pop(0)
            self.done = False
        else:
            self.done = True
        self._key = None

    def read_value(self, symbol):
        if isinstance(self._current, dict):
            if self._key not in self._current:
                # Use the default value
                return symbol.get_default()
            else:
                return self._current[self._key]
        else:
            # If we aren't in a dict or a list then this must be a schema which
            # just has a single basic type
            return self._current

    def _push(self, symbol=None):
        self._stack.append((self._current, self._key))
        if isinstance(self._current, dict) and self._key is not None:
            if self._key not in self._current:
                self._current = symbol.get_default()
            else:
                # self._current = self._current.pop(self._key)
                self._current = self._current[self._key]

    def _pop(self):
        self._current, self._key = self._stack.pop()

    def configure(self, schema, named_schemas):
        self._parser = Parser(schema, named_schemas, self.do_action)

    def do_action(self, action):
        if isinstance(action, RecordStart):
            self._push(action)
        elif isinstance(action, RecordEnd):
            self._pop()
        elif isinstance(action, FieldStart):
            self.read_object_key(action.field_name)
        elif isinstance(action, FieldEnd) or isinstance(action, UnionEnd):
            # TODO: Do we need a FieldEnd and UnionEnd symbol?
            pass
        else:
            raise Exception(f"cannot handle: {action}")

    def drain(self):
        self._parser.drain_actions()
        if self._json_data:
            self._current = self._json_data.pop(0)
            self._key = None
        else:
            self.done = True

    def read_null(self):
        symbol = self._parser.advance(Null())
        return self.read_value(symbol)

    def read_boolean(self):
        symbol = self._parser.advance(Boolean())
        return self.read_value(symbol)

    def read_utf8(self):
        symbol = self._parser.advance(String())
        if self._parser.stack[-1] == MapKeyMarker():
            self._parser.advance(MapKeyMarker())
            for key in self._current:
                self._key = key
                break
            return self._key
        else:
            return self.read_value(symbol)

    def read_bytes(self):
        symbol = self._parser.advance(Bytes())
        return self.read_value(symbol).encode("iso-8859-1")

    def read_int(self):
        symbol = self._parser.advance(Int())
        return self.read_value(symbol)

    def read_long(self):
        symbol = self._parser.advance(Long())
        return self.read_value(symbol)

    def read_float(self):
        symbol = self._parser.advance(Float())
        return self.read_value(symbol)

    def read_double(self):
        symbol = self._parser.advance(Double())
        return self.read_value(symbol)

    def read_enum(self):
        symbol = self._parser.advance(Enum())
        enum_labels = self._parser.pop_symbol()  # pop the enumlabels
        # TODO: Should we verify the value is one of the symbols?
        label = self.read_value(symbol)
        return enum_labels.labels.index(label)

    def read_fixed(self, size):
        symbol = self._parser.advance(Fixed())
        return self.read_value(symbol).encode("iso-8859-1")

    def read_map_start(self):
        symbol = self._parser.advance(MapStart())
        self._push(symbol)

    def read_object_key(self, key):
        self._key = key

    def iter_map(self):
        while len(self._current) > 0:
            yield
            del self._current[self._key]

    def read_map_end(self):
        self._parser.advance(MapEnd())
        self._pop()

    def read_array_start(self):
        symbol = self._parser.advance(ArrayStart())
        self._push(symbol)
        self._key = None

    def read_array_end(self):
        self._parser.advance(ArrayEnd())
        self._pop()

    def iter_array(self):
        while len(self._current) > 0:
            self._push()
            self._current = self._current.pop(0)
            yield
            self._pop()
            self._parser.advance(ItemEnd())

    def read_index(self):
        self._parser.advance(Union())
        alternative_symbol = self._parser.pop_symbol()

        # TODO: Try to clean this up.
        # A JSON union is encoded like this: {"union_field": {int: 32}} and so
        # what we are doing is trying to change that into {"union_field": 32}
        # before eventually reading the value of "union_field"
        if self._key is None:
            # If self._key is None, self._current is an item in an array
            if self._current is None:
                label = "null"
            else:
                label, data = self._current.popitem()
                self._current = data
                # TODO: Do we need to do this?
                self._parser.push_symbol(UnionEnd())
        else:
            # self._current is a JSON object and self._key should be the name
            # of the union field
            if self._key not in self._current:
                self._current[self._key] = {
                    alternative_symbol.labels[0]: alternative_symbol.get_default()
                }

            if self._current[self._key] is None:
                label = "null"
            else:
                label, data = self._current[self._key].popitem()
                self._current[self._key] = data
                # TODO: Do we need to do this?
                self._parser.push_symbol(UnionEnd())

        index = alternative_symbol.labels.index(label)
        symbol = alternative_symbol.get_symbol(index)
        self._parser.push_symbol(symbol)
        return index FINGERPRINT_ALGORITHMS = _schema.FINGERPRINT_ALGORITHMS class SchemaResolutionError(Exception):
    pass MAGIC = b"Obj" + chr(VERSION).encode() SYNC_SIZE = 16 HEADER_SCHEMA = {
    "type": "record",
    "name": "org.apache.avro.file.Header",
    "fields": [
        {
            "name": "magic",
            "type": {"type": "fixed", "name": "magic", "size": len(MAGIC)},
        },
        {"name": "meta", "type": {"type": "map", "values": "bytes"}},
        {"name": "sync", "type": {"type": "fixed", "name": "sync", "size": SYNC_SIZE}},
    ],
} def missing_codec_lib(codec, library):
    def missing(fo):
        raise ValueError(
            f"{codec} codec is supported but you need to install {library}"
        )

    return missing MCS_PER_HOUR = MCS_PER_MINUTE * 60 MCS_PER_MINUTE = MCS_PER_SECOND * 60 MCS_PER_SECOND = 1000000 MLS_PER_HOUR = MLS_PER_MINUTE * 60 MLS_PER_MINUTE = MLS_PER_SECOND * 60 MLS_PER_SECOND = 1000 DAYS_SHIFT = datetime.date(1970, 1, 1).toordinal() NAMED_TYPES = {"record", "enum", "fixed", "error"}
from io import BytesIO
from struct import error as StructError
from datetime import datetime, time, date, timezone, timedelta
from decimal import Context
from uuid import UUID
from .io.binary_decoder import BinaryDecoder
from .io.json_decoder import AvroJSONDecoder
from .schema import extract_record_type, extract_logical_type, parse_schema
from ._read_common import (
    SchemaResolutionError,
    MAGIC,
    SYNC_SIZE,
    HEADER_SCHEMA,
    missing_codec_lib,
)
from .const import (
    MCS_PER_HOUR,
    MCS_PER_MINUTE,
    MCS_PER_SECOND,
    MLS_PER_HOUR,
    MLS_PER_MINUTE,
    MLS_PER_SECOND,
    DAYS_SHIFT,
    NAMED_TYPES,
)
import bz2
import lzma
import zlib
import json
    import snappy
    import zstandard as zstd
    import lz4.block
        import warnings
    fo: file-like
        Input stream
    reader_schema: dict, optional
        Reader schema
    return_record_name: bool, optional
        If true, when reading a union of records, the result will be a tuple
        where the first value is the name of the record and the second value is
        the record itself


    Example::

        from fastavro import block_reader
        with open('some-file.avro', 'rb') as fo:
            avro_reader = block_reader(fo)
            for block in avro_reader:
                process_block(block)

    .. attribute:: metadata

        Key-value pairs in the header metadata

    .. attribute:: codec

        The codec used when writing

    .. attribute:: writer_schema

        The schema used when writing

    .. attribute:: reader_schema

        The schema used when reading (if provided)
    """

    def __init__(self, fo, reader_schema=None, return_record_name=False):
        file_reader.__init__(self, fo, reader_schema, return_record_name)

        self._read_header()

        self._elems = _iter_avro_blocks(
            self.decoder,
            self._header,
            self.codec,
            self.writer_schema,
            self._named_schemas,
            self.reader_schema,
            self.return_record_name,
        )


def schemaless_reader(fo, writer_schema, reader_schema=None, return_record_name=False):
    """Reads a single record writen using the
    :meth:`~fastavro._write_py.schemaless_writer`

    Parameters
    ----------
    fo: file-like
        Input stream
    writer_schema: dict
        Schema used when calling schemaless_writer
    reader_schema: dict, optional
        If the schema has changed since being written then the new schema can
        be given to allow for schema migration
    return_record_name: bool, optional
        If true, when reading a union of records, the result will be a tuple
        where the first value is the name of the record and the second value is
        the record itself


    Example::

        parsed_schema = fastavro.parse_schema(schema)
        with open('file', 'rb') as fp:
            record = fastavro.schemaless_reader(fp, parsed_schema)

    Note: The ``schemaless_reader`` can only read a single record.
    """
    if writer_schema == reader_schema:
        # No need for the reader schema if they are the same
        reader_schema = None

    named_schemas = {}
    writer_schema = parse_schema(writer_schema, named_schemas)

    if reader_schema:
        reader_schema = parse_schema(reader_schema)

    decoder = BinaryDecoder(fo)

    return read_data(
        decoder,
        writer_schema,
        named_schemas,
        reader_schema,
        return_record_name,
    )


def is_avro(path_or_buffer):
    """Return True if path (or buffer) points to an Avro file. This will only
    work for avro files that contain the normal avro schema header like those
    create from :func:`~fastavro._write_py.writer`. This function is not intended
    to be used with binary data created from
    :func:`~fastavro._write_py.schemaless_writer` since that does not include the
    avro header.

    Parameters
    ----------
    path_or_buffer: path to file or file-like object
        Path to file
    """
    if isinstance(path_or_buffer, str):
        fp = open(path_or_buffer, "rb")
        close = True
    else:
        fp = path_or_buffer
        close = False

    try:

        header = fp.read(len(MAGIC))