This is an automated email from the ASF dual-hosted git repository.
kojiromike pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/avro.git
The following commit(s) were added to refs/heads/master by this push:
new 6c4e165 AVRO-2921: Type Fixes for avro.io (#1264)
6c4e165 is described below
commit 6c4e165584431aeb21ff8f84fe1cefc5c398fa8e
Author: Michael A. Smith <[email protected]>
AuthorDate: Wed Jun 23 08:04:34 2021 -0400
AVRO-2921: Type Fixes for avro.io (#1264)
---
lang/py/avro/datafile.py | 22 +-
lang/py/avro/errors.py | 10 +-
lang/py/avro/io.py | 587 ++++++++++++++++++++++------------------
lang/py/avro/protocol.py | 7 +-
lang/py/avro/schema.py | 70 +++--
lang/py/avro/test/test_bench.py | 4 +-
lang/py/avro/test/test_io.py | 74 ++---
lang/py/avro/utils.py | 6 +
8 files changed, 438 insertions(+), 342 deletions(-)
diff --git a/lang/py/avro/datafile.py b/lang/py/avro/datafile.py
index 0e0bd5d..87383ff 100644
--- a/lang/py/avro/datafile.py
+++ b/lang/py/avro/datafile.py
@@ -25,13 +25,13 @@
https://avro.apache.org/docs/current/spec.html#Object+Container+Files
import io
import json
from types import TracebackType
-from typing import BinaryIO, MutableMapping, Optional, Type
+from typing import BinaryIO, MutableMapping, Optional, Type, cast
import avro.codecs
import avro.errors
import avro.io
import avro.schema
-import avro.utils
+from avro.utils import TypedDict, randbytes
VERSION = 1
MAGIC = bytes(b"Obj" + bytearray([VERSION]))
@@ -60,6 +60,12 @@ CODEC_KEY = "avro.codec"
SCHEMA_KEY = "avro.schema"
+class HeaderType(TypedDict):
+ magic: bytes
+ meta: MutableMapping[str, bytes]
+ sync: bytes
+
+
class _DataFileMetadata:
"""
Mixin for meta properties.
@@ -183,7 +189,7 @@ class DataFileWriter(_DataFileMetadata):
writer.seek(0, 2)
self._header_written = True
return
- self.sync_marker = avro.utils.randbytes(16)
+ self.sync_marker = randbytes(16)
self.codec = codec
self.schema = str(writers_schema)
self.datum_writer.writers_schema = writers_schema
@@ -358,16 +364,10 @@ class DataFileReader(_DataFileMetadata):
self.reader.seek(0, 0)
# read header into a dict
- header = self.datum_reader.read_data(META_SCHEMA, META_SCHEMA,
self.raw_decoder)
-
- # check magic number
+ header = cast(HeaderType, self.datum_reader.read_data(META_SCHEMA,
META_SCHEMA, self.raw_decoder))
if header.get("magic") != MAGIC:
raise avro.errors.AvroException(f"Not an Avro data file:
{header.get('magic')!r} doesn't match {MAGIC!r}.")
-
- # set metadata
self._meta = header["meta"]
-
- # set sync marker
self.sync_marker = header["sync"]
def _read_block_header(self) -> None:
@@ -393,6 +393,8 @@ class DataFileReader(_DataFileMetadata):
raise StopIteration
self._read_block_header()
+ if self.datum_decoder is None:
+ raise avro.errors.DataFileException("DataFile is not ready to read
because it has no decoder")
datum = self.datum_reader.read(self.datum_decoder)
self.block_count -= 1
return datum
diff --git a/lang/py/avro/errors.py b/lang/py/avro/errors.py
index b613e55..1723b6c 100644
--- a/lang/py/avro/errors.py
+++ b/lang/py/avro/errors.py
@@ -56,7 +56,11 @@ class AvroTypeException(AvroException):
expected_schema, datum = args[:2]
except (IndexError, ValueError):
return super().__init__(*args)
- return super().__init__(f"The datum {datum} is not an example of the
schema {_safe_pretty(expected_schema)}")
+ return super().__init__(f"The datum {datum} of the type {type(datum)}
is not an example of the schema {_safe_pretty(expected_schema)}")
+
+
+class InvalidDefaultException(AvroTypeException):
+ """Raised when a default value isn't a suitable type for the schema."""
class AvroOutOfScaleException(AvroTypeException):
@@ -81,6 +85,10 @@ class DataFileException(AvroException):
"""Raised when there's a problem reading or writing file object
containers."""
+class IONotReadyException(AvroException):
+ """Raised when attempting an avro operation on an io object that isn't
fully initialized."""
+
+
class AvroRemoteException(AvroException):
"""Raised when an error message is sent by an Avro requestor or
responder."""
diff --git a/lang/py/avro/io.py b/lang/py/avro/io.py
index 0978b3a..161f739 100644
--- a/lang/py/avro/io.py
+++ b/lang/py/avro/io.py
@@ -87,18 +87,25 @@ in that datum, if there are any.
import collections
import datetime
import decimal
-import json
import struct
+import warnings
+from typing import (
+ BinaryIO,
+ Deque,
+ Generator,
+ Iterable,
+ List,
+ Mapping,
+ Optional,
+ Sequence,
+ Union,
+)
import avro.constants
import avro.errors
+import avro.schema
import avro.timezones
-#
-# Constants
-#
-
-
# 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
@@ -106,16 +113,12 @@ 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
-
-#
-# Validate
-#
-
-
ValidationNode = collections.namedtuple("ValidationNode", ["schema", "datum",
"name"])
+ValidationNodeGeneratorType = Generator[ValidationNode, None, None]
+JsonScalarFieldType = Union[None, bool, str, int, float]
-def validate(expected_schema, datum, raise_on_error=False):
+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
@@ -130,38 +133,27 @@ def validate(expected_schema, datum,
raise_on_error=False):
: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()
- nodes.append(ValidationNode(expected_schema, datum,
getattr(expected_schema, "name", None)))
+ 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
- # if current_node.schema.type in {'array', 'map', 'record'}:
validated_schema = current_node.schema.validate(current_node.datum)
- if validated_schema:
- valid_node = ValidationNode(validated_schema, current_node.datum,
current_node.name)
- else:
- valid_node = None
- # else:
- # valid_node = _validate_node(current_node)
-
- if valid_node is not None:
- # if there are children of this node to append, do so.
- for child_node in _iterate_node(valid_node):
- nodes.append(child_node)
- else:
- # the current node was not valid.
+ 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.datum)
- else:
- # preserve the prior validation behavior of returning false
when there are problems.
- return False
+ 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):
+def _iterate_node(node: ValidationNode) -> ValidationNodeGeneratorType:
for item in _ITERATORS.get(node.schema.type, _default_iterator)(node):
yield ValidationNode(*item)
@@ -171,7 +163,7 @@ def _iterate_node(node):
#############
-def _default_iterator(_):
+def _default_iterator(_) -> ValidationNodeGeneratorType:
"""Immediately raise StopIteration.
This exists to prevent problems with iteration over unsupported container
types.
@@ -179,26 +171,23 @@ def _default_iterator(_):
yield from ()
-def _record_iterator(node):
+def _record_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
"""Yield each child node of the provided record node."""
- schema, datum, name = node
- for field in schema.fields:
- yield ValidationNode(field.type, datum.get(field.name), field.name) #
type: ignore
+ schema, datum, _ = node
+ return (ValidationNode(field.type, datum.get(field.name), field.name) for
field in schema.fields)
-def _array_iterator(node):
+def _array_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
"""Yield each child node of the provided array node."""
schema, datum, name = node
- for item in datum: # type: ignore
- yield ValidationNode(schema.items, item, name)
+ return (ValidationNode(schema.items, item, name) for item in datum)
-def _map_iterator(node):
+def _map_iterator(node: ValidationNode) -> ValidationNodeGeneratorType:
"""Yield each child node of the provided map node."""
schema, datum, _ = node
child_schema = schema.values
- for child_name, child_datum in datum.items(): # type: ignore
- yield ValidationNode(child_schema, child_datum, child_name)
+ return (ValidationNode(child_schema, child_datum, child_name) for
child_name, child_datum in datum.items())
_ITERATORS = {
@@ -217,41 +206,44 @@ _ITERATORS["error"] = _ITERATORS["request"] =
_ITERATORS["record"]
class BinaryDecoder:
"""Read leaf values."""
- def __init__(self, reader):
+ _reader: BinaryIO
+
+ def __init__(self, reader: BinaryIO) -> None:
"""
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)
+ @property
+ def reader(self) -> BinaryIO:
+ return self._reader
- def read(self, n):
+ def read(self, n: int) -> bytes:
"""
Read n bytes.
"""
return self.reader.read(n)
- def read_null(self):
+ def read_null(self) -> None:
"""
null is written as zero bytes
"""
return None
- def read_boolean(self):
+ 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):
+ 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):
+ def read_long(self) -> int:
"""
int and long values are written using variable-length, zig-zag coding.
"""
@@ -265,23 +257,23 @@ class BinaryDecoder:
datum = (n >> 1) ^ -(n & 1)
return datum
- def read_float(self):
+ 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 floatToIntBits and then encoded in little-endian format.
"""
- return STRUCT_FLOAT.unpack(self.read(4))[0]
+ return float(STRUCT_FLOAT.unpack(self.read(4))[0])
- def read_double(self):
+ 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 doubleToLongBits and then encoded in little-endian format.
"""
- return STRUCT_DOUBLE.unpack(self.read(8))[0]
+ return float(STRUCT_DOUBLE.unpack(self.read(8))[0])
- def read_decimal_from_bytes(self, precision, scale):
+ 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.
@@ -289,7 +281,7 @@ class BinaryDecoder:
size = self.read_long()
return self.read_decimal_from_fixed(precision, scale, size)
- def read_decimal_from_fixed(self, 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.
@@ -318,20 +310,20 @@ class BinaryDecoder:
decimal.getcontext().prec = original_prec
return scaled_datum
- def read_bytes(self):
+ 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):
+ 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):
+ def read_date_from_int(self) -> datetime.date:
"""
int is decoded as python date object.
int stores the number of days from
@@ -340,7 +332,7 @@ class BinaryDecoder:
days_since_epoch = self.read_int()
return datetime.date(1970, 1, 1) + datetime.timedelta(days_since_epoch)
- def _build_time_object(self, value, scale_to_micro):
+ 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)
@@ -349,7 +341,7 @@ class BinaryDecoder:
return datetime.time(hour=hours, minute=minutes, second=seconds,
microsecond=microseconds)
- def read_time_millis_from_int(self):
+ 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.
@@ -357,7 +349,7 @@ class BinaryDecoder:
milliseconds = self.read_int()
return self._build_time_object(milliseconds, 1000)
- def read_time_micros_from_long(self):
+ 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.
@@ -365,7 +357,7 @@ class BinaryDecoder:
microseconds = self.read_long()
return self._build_time_object(microseconds, 1)
- def read_timestamp_millis_from_long(self):
+ 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.
@@ -375,7 +367,7 @@ class BinaryDecoder:
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):
+ 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.
@@ -385,72 +377,75 @@ class BinaryDecoder:
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):
+ def skip_null(self) -> None:
pass
- def skip_boolean(self):
+ def skip_boolean(self) -> None:
self.skip(1)
- def skip_int(self):
+ def skip_int(self) -> None:
self.skip_long()
- def skip_long(self):
+ def skip_long(self) -> None:
b = ord(self.read(1))
while (b & 0x80) != 0:
b = ord(self.read(1))
- def skip_float(self):
+ def skip_float(self) -> None:
self.skip(4)
- def skip_double(self):
+ def skip_double(self) -> None:
self.skip(8)
- def skip_bytes(self):
+ def skip_bytes(self) -> None:
self.skip(self.read_long())
- def skip_utf8(self):
+ def skip_utf8(self) -> None:
self.skip_bytes()
- def skip(self, n):
+ def skip(self, n: int) -> None:
self.reader.seek(self.reader.tell() + n)
class BinaryEncoder:
"""Write leaf values."""
- def __init__(self, writer):
+ _writer: BinaryIO
+
+ def __init__(self, writer: BinaryIO) -> None:
"""
writer is a Python object on which we can call write.
"""
self._writer = writer
- # read-only properties
- writer = property(lambda self: self._writer)
+ @property
+ def writer(self) -> BinaryIO:
+ return self._writer
- def write(self, datum):
+ def write(self, datum: bytes) -> None:
"""Write an arbitrary datum."""
self.writer.write(datum)
- def write_null(self, datum):
+ def write_null(self, datum: None) -> None:
"""
null is written as zero bytes
"""
pass
- def write_boolean(self, datum):
+ 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):
+ 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):
+ def write_long(self, datum: int) -> None:
"""
int and long values are written using variable-length, zig-zag coding.
"""
@@ -460,7 +455,7 @@ class BinaryEncoder:
datum >>= 7
self.write(bytearray([datum]))
- def write_float(self, 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
@@ -468,7 +463,7 @@ class BinaryEncoder:
"""
self.write(STRUCT_FLOAT.pack(datum))
- def write_double(self, 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
@@ -476,7 +471,7 @@ class BinaryEncoder:
"""
self.write(STRUCT_DOUBLE.pack(datum))
- def write_decimal_bytes(self, datum, scale):
+ 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.
@@ -503,7 +498,7 @@ class BinaryEncoder:
bits_to_write = packed_bits >> (8 * index)
self.write(bytearray([bits_to_write & 0xFF]))
- def write_decimal_fixed(self, datum, scale, size):
+ def write_decimal_fixed(self, datum: decimal.Decimal, scale: int, size:
int) -> None:
"""
Decimal in fixed are encoded as size of fixed bytes.
"""
@@ -544,22 +539,21 @@ class BinaryEncoder:
bits_to_write = unscaled_datum >> (8 * index)
self.write(bytearray([bits_to_write & 0xFF]))
- def write_bytes(self, datum):
+ 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):
+ 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.
"""
- datum = datum.encode("utf-8")
- self.write_bytes(datum)
+ self.write_bytes(datum.encode("utf-8"))
- def write_date_int(self, datum):
+ def write_date_int(self, datum: datetime.date) -> None:
"""
Encode python date object as int.
It stores the number of days from
@@ -568,7 +562,7 @@ class BinaryEncoder:
delta_date = datum - datetime.date(1970, 1, 1)
self.write_int(delta_date.days)
- def write_time_millis_int(self, datum):
+ 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
@@ -576,7 +570,7 @@ class BinaryEncoder:
milliseconds = datum.hour * 3600000 + datum.minute * 60000 +
datum.second * 1000 + datum.microsecond // 1000
self.write_int(milliseconds)
- def write_time_micros_long(self, datum):
+ 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
@@ -584,10 +578,10 @@ class BinaryEncoder:
microseconds = datum.hour * 3600000000 + datum.minute * 60000000 +
datum.second * 1000000 + datum.microsecond
self.write_long(microseconds)
- def _timedelta_total_microseconds(self, timedelta):
- return timedelta.microseconds + (timedelta.seconds + timedelta.days *
24 * 3600) * 10 ** 6
+ def _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):
+ 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.
@@ -597,7 +591,7 @@ class BinaryEncoder:
milliseconds = self._timedelta_total_microseconds(timedelta) // 1000
self.write_long(milliseconds)
- def write_timestamp_micros_long(self, datum):
+ 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.
@@ -614,7 +608,10 @@ class BinaryEncoder:
class DatumReader:
"""Deserialize Avro-encoded data into a Python data structure."""
- def __init__(self, writers_schema=None, readers_schema=None):
+ _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
@@ -623,139 +620,150 @@ class DatumReader:
self._writers_schema = writers_schema
self._readers_schema = readers_schema
- # read/write properties
- def set_writers_schema(self, 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
- writers_schema = property(lambda self: self._writers_schema,
set_writers_schema)
+ @property
+ def readers_schema(self) -> Optional[avro.schema.Schema]:
+ return self._readers_schema
- def set_readers_schema(self, readers_schema):
+ @readers_schema.setter
+ def readers_schema(self, readers_schema: avro.schema.Schema) -> None:
self._readers_schema = readers_schema
- readers_schema = property(lambda self: self._readers_schema,
set_readers_schema)
-
- def read(self, decoder):
+ 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, 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):
- fail_msg = "Schemas do not match."
- raise avro.errors.SchemaResolutionException(fail_msg,
writers_schema, readers_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 writers_schema.type in ["union", "error_union"]:
+ if isinstance(writers_schema, avro.schema.UnionSchema) and
isinstance(readers_schema, avro.schema.UnionSchema):
return self.read_union(writers_schema, readers_schema, decoder)
- if readers_schema.type in ["union", "error_union"]:
+ 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.
- fail_msg = "Schemas do not match."
- raise avro.errors.SchemaResolutionException(fail_msg,
writers_schema, readers_schema)
+ raise avro.errors.SchemaResolutionException("Schemas do not
match.", writers_schema, readers_schema)
if writers_schema.type == "null":
- return decoder.read_null()
- elif writers_schema.type == "boolean":
+ return None
+ if writers_schema.type == "boolean":
return decoder.read_boolean()
- elif writers_schema.type == "string":
+ if writers_schema.type == "string":
return decoder.read_utf8()
- elif writers_schema.type == "int":
+ 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()
- elif writers_schema.type == "long":
+ if writers_schema.type == "long":
if logical_type == avro.constants.TIME_MICROS:
return decoder.read_time_micros_from_long()
- elif logical_type == avro.constants.TIMESTAMP_MILLIS:
+ if logical_type == avro.constants.TIMESTAMP_MILLIS:
return decoder.read_timestamp_millis_from_long()
- elif logical_type == avro.constants.TIMESTAMP_MICROS:
+ if logical_type == avro.constants.TIMESTAMP_MICROS:
return decoder.read_timestamp_micros_from_long()
- else:
- return decoder.read_long()
- elif writers_schema.type == "float":
+ return decoder.read_long()
+ if writers_schema.type == "float":
return decoder.read_float()
- elif writers_schema.type == "double":
+ if writers_schema.type == "double":
return decoder.read_double()
- elif writers_schema.type == "bytes":
+ if writers_schema.type == "bytes":
if logical_type == "decimal":
- return decoder.read_decimal_from_bytes(
- writers_schema.get_prop("precision"),
- writers_schema.get_prop("scale"),
- )
- else:
- return decoder.read_bytes()
- elif writers_schema.type == "fixed":
+ 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 positive 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":
- return decoder.read_decimal_from_fixed(
- writers_schema.get_prop("precision"),
- writers_schema.get_prop("scale"),
- writers_schema.size,
- )
+ 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 positive 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)
- elif writers_schema.type == "enum":
+ if isinstance(writers_schema, avro.schema.EnumSchema) and
isinstance(readers_schema, avro.schema.EnumSchema):
return self.read_enum(writers_schema, readers_schema, decoder)
- elif writers_schema.type == "array":
+ if isinstance(writers_schema, avro.schema.ArraySchema) and
isinstance(readers_schema, avro.schema.ArraySchema):
return self.read_array(writers_schema, readers_schema, decoder)
- elif writers_schema.type == "map":
+ if isinstance(writers_schema, avro.schema.MapSchema) and
isinstance(readers_schema, avro.schema.MapSchema):
return self.read_map(writers_schema, readers_schema, decoder)
- elif writers_schema.type in ["record", "error", "request"]:
+ 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)
- else:
- raise avro.errors.AvroException(f"Cannot read unknown schema type:
{writers_schema.type}")
+ raise avro.errors.AvroException(f"Cannot read unknown schema type:
{writers_schema.type}")
- def skip_data(self, writers_schema, decoder):
+ def skip_data(self, writers_schema: avro.schema.Schema, decoder:
BinaryDecoder) -> None:
if writers_schema.type == "null":
return decoder.skip_null()
- elif writers_schema.type == "boolean":
+ if writers_schema.type == "boolean":
return decoder.skip_boolean()
- elif writers_schema.type == "string":
+ if writers_schema.type == "string":
return decoder.skip_utf8()
- elif writers_schema.type == "int":
+ if writers_schema.type == "int":
return decoder.skip_int()
- elif writers_schema.type == "long":
+ if writers_schema.type == "long":
return decoder.skip_long()
- elif writers_schema.type == "float":
+ if writers_schema.type == "float":
return decoder.skip_float()
- elif writers_schema.type == "double":
+ if writers_schema.type == "double":
return decoder.skip_double()
- elif writers_schema.type == "bytes":
+ if writers_schema.type == "bytes":
return decoder.skip_bytes()
- elif writers_schema.type == "fixed":
+ if isinstance(writers_schema, avro.schema.FixedSchema):
return self.skip_fixed(writers_schema, decoder)
- elif writers_schema.type == "enum":
+ if isinstance(writers_schema, avro.schema.EnumSchema):
return self.skip_enum(writers_schema, decoder)
- elif writers_schema.type == "array":
+ if isinstance(writers_schema, avro.schema.ArraySchema):
return self.skip_array(writers_schema, decoder)
- elif writers_schema.type == "map":
+ if isinstance(writers_schema, avro.schema.MapSchema):
return self.skip_map(writers_schema, decoder)
- elif writers_schema.type in ["union", "error_union"]:
+ if isinstance(writers_schema, avro.schema.UnionSchema):
return self.skip_union(writers_schema, decoder)
- elif writers_schema.type in ["record", "error", "request"]:
+ if isinstance(writers_schema, avro.schema.RecordSchema):
return self.skip_record(writers_schema, decoder)
- else:
- raise avro.errors.AvroException(f"Unknown schema type:
{writers_schema.type}")
+ raise avro.errors.AvroException(f"Unknown schema type:
{writers_schema.type}")
- def read_fixed(self, writers_schema, readers_schema, decoder):
+ 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, decoder):
+ def skip_fixed(self, writers_schema: avro.schema.FixedSchema, decoder:
BinaryDecoder) -> None:
return decoder.skip(writers_schema.size)
- def read_enum(self, writers_schema, readers_schema, decoder):
+ 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.
@@ -764,7 +772,7 @@ class DatumReader:
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_symbole} for enum with
{len(writers_schema.symbols)} symbols", writers_schema, readers_schema
+ 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]
@@ -774,10 +782,10 @@ class DatumReader:
return read_symbol
- def skip_enum(self, writers_schema, decoder):
+ def skip_enum(self, writers_schema: avro.schema.EnumSchema, decoder:
BinaryDecoder) -> None:
return decoder.skip_int()
- def read_array(self, writers_schema, readers_schema, decoder):
+ 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.
@@ -803,7 +811,7 @@ class DatumReader:
block_count = decoder.read_long()
return read_items
- def skip_array(self, writers_schema, decoder):
+ 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:
@@ -814,7 +822,7 @@ class DatumReader:
self.skip_data(writers_schema.items, decoder)
block_count = decoder.read_long()
- def read_map(self, writers_schema, readers_schema, decoder):
+ 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.
@@ -841,7 +849,7 @@ class DatumReader:
block_count = decoder.read_long()
return read_items
- def skip_map(self, writers_schema, decoder):
+ 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:
@@ -853,7 +861,7 @@ class DatumReader:
self.skip_data(writers_schema.values, decoder)
block_count = decoder.read_long()
- def read_union(self, writers_schema, readers_schema, decoder):
+ 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.
@@ -870,7 +878,7 @@ class DatumReader:
# read data
return self.read_data(selected_writers_schema, readers_schema, decoder)
- def skip_union(self, writers_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(
@@ -878,7 +886,9 @@ class DatumReader:
)
return self.skip_data(writers_schema.schemas[index_of_schema], decoder)
- def read_record(self, writers_schema, readers_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
@@ -920,41 +930,51 @@ class DatumReader:
read_record[field.name] = field_val
return read_record
- def skip_record(self, writers_schema, decoder):
+ 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, default_value):
+ def _read_default_value(self, field_schema: avro.schema.Schema,
default_value: object) -> object:
"""
Basically a JSON Decoder?
"""
if field_schema.type == "null":
- return None
- elif field_schema.type == "boolean":
+ if default_value is None:
+ return None
+ raise avro.errors.InvalidDefaultException(field_schema,
default_value)
+ if field_schema.type == "boolean":
return bool(default_value)
- elif field_schema.type == "int":
- return int(default_value)
- elif field_schema.type == "long":
- return int(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"]:
+ 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)
- elif field_schema.type == "record":
+ 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)
@@ -963,98 +983,143 @@ class DatumReader:
field_val = self._read_default_value(field.type, json_val)
read_record[field.name] = field_val
return read_record
- else:
- raise avro.errors.AvroException(f"Unknown type:
{field_schema.type}")
+ raise avro.errors.AvroException(f"Unknown type: {field_schema.type}")
class DatumWriter:
"""DatumWriter for generic python objects."""
- def __init__(self, writers_schema=None):
- self._writers_schema = writers_schema
+ _writers_schema: Optional[avro.schema.Schema]
- # read/write properties
- def set_writers_schema(self, writers_schema):
+ def __init__(self, writers_schema: Optional[avro.schema.Schema] = None) ->
None:
self._writers_schema = writers_schema
- writers_schema = property(lambda self: self._writers_schema,
set_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, encoder):
+ 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, 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":
- encoder.write_null(datum)
- elif writers_schema.type == "boolean":
- encoder.write_boolean(datum)
- elif writers_schema.type == "string":
- encoder.write_utf8(datum)
- elif writers_schema.type == "int":
+ if 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:
- encoder.write_date_int(datum)
+ 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:
- encoder.write_time_millis_int(datum)
- else:
- encoder.write_int(datum)
- elif writers_schema.type == "long":
+ 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:
- encoder.write_time_micros_long(datum)
+ 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:
- encoder.write_timestamp_millis_long(datum)
+ 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:
- encoder.write_timestamp_micros_long(datum)
- else:
- encoder.write_long(datum)
- elif writers_schema.type == "float":
- encoder.write_float(datum)
- elif writers_schema.type == "double":
- encoder.write_double(datum)
- elif writers_schema.type == "bytes":
+ if 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":
- encoder.write_decimal_bytes(datum,
writers_schema.get_prop("scale"))
- else:
- encoder.write_bytes(datum)
- elif writers_schema.type == "fixed":
+ 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 positive 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":
- encoder.write_decimal_fixed(
- datum,
- writers_schema.get_prop("scale"),
- writers_schema.get_prop("size"),
- )
- else:
- self.write_fixed(writers_schema, datum, encoder)
- elif writers_schema.type == "enum":
- self.write_enum(writers_schema, datum, encoder)
- elif writers_schema.type == "array":
- self.write_array(writers_schema, datum, encoder)
- elif writers_schema.type == "map":
- self.write_map(writers_schema, datum, encoder)
- elif writers_schema.type in ["union", "error_union"]:
- self.write_union(writers_schema, datum, encoder)
- elif writers_schema.type in ["record", "error", "request"]:
- self.write_record(writers_schema, datum, encoder)
- else:
- raise avro.errors.AvroException(f"Unknown type:
{writers_schema.type}")
+ 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 positive 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, datum, encoder):
+ 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.
"""
- encoder.write(datum)
+ return encoder.write(datum)
- def write_enum(self, writers_schema, datum, encoder):
+ 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)
- encoder.write_int(index_of_datum)
+ return encoder.write_int(index_of_datum)
- def write_array(self, writers_schema, datum, encoder):
+ def write_array(self, writers_schema: avro.schema.ArraySchema, datum:
Sequence[object], encoder: BinaryEncoder) -> None:
"""
Arrays are encoded as a series of blocks.
@@ -1073,9 +1138,9 @@ class DatumWriter:
encoder.write_long(len(datum))
for item in datum:
self.write_data(writers_schema.items, item, encoder)
- encoder.write_long(0)
+ return encoder.write_long(0)
- def write_map(self, writers_schema, datum, encoder):
+ def write_map(self, writers_schema: avro.schema.MapSchema, datum:
Mapping[str, object], encoder: BinaryEncoder) -> None:
"""
Maps are encoded as a series of blocks.
@@ -1095,9 +1160,9 @@ class DatumWriter:
for key, val in datum.items():
encoder.write_utf8(key)
self.write_data(writers_schema.values, val, encoder)
- encoder.write_long(0)
+ return encoder.write_long(0)
- def write_union(self, writers_schema, datum, encoder):
+ 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.
@@ -1113,9 +1178,9 @@ class DatumWriter:
# write data
encoder.write_long(index_of_schema)
- self.write_data(writers_schema.schemas[index_of_schema], datum,
encoder)
+ return self.write_data(writers_schema.schemas[index_of_schema], datum,
encoder)
- def write_record(self, writers_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
diff --git a/lang/py/avro/protocol.py b/lang/py/avro/protocol.py
index 4c27f88..e345bae 100644
--- a/lang/py/avro/protocol.py
+++ b/lang/py/avro/protocol.py
@@ -25,21 +25,16 @@
https://avro.apache.org/docs/current/spec.html#Protocol+Declaration
import hashlib
import json
-import sys
from typing import Mapping, Optional, Sequence, Union, cast
import avro.errors
import avro.name
import avro.schema
+from avro.utils import TypedDict
# TODO(hammer): confirmed 'fixed' with Doug
VALID_TYPE_SCHEMA_TYPES = ("enum", "record", "error", "fixed")
-if sys.version_info >= (3, 8):
- from typing import TypedDict
-else:
- from typing_extensions import TypedDict
-
class MessageObject(TypedDict, total=False):
request: Sequence[Mapping[str, object]]
diff --git a/lang/py/avro/schema.py b/lang/py/avro/schema.py
index cb61dad..1f73931 100644
--- a/lang/py/avro/schema.py
+++ b/lang/py/avro/schema.py
@@ -554,14 +554,14 @@ class FixedDecimalSchema(FixedSchema,
DecimalLogicalSchema):
class EnumSchema(EqualByPropsMixin, NamedSchema):
def __init__(
self,
- name,
- namespace,
- symbols,
- names=None,
- doc=None,
- other_props=None,
- validate_enum_symbols=True,
- ):
+ name: str,
+ namespace: str,
+ symbols: Sequence[str],
+ names: Optional[avro.name.Names] = None,
+ doc: Optional[str] = None,
+ other_props: Optional[Mapping[str, object]] = None,
+ validate_enum_symbols: bool = True,
+ ) -> None:
"""
@arg validate_enum_symbols: If False, will allow enum symbols that are
not valid Avro names.
"""
@@ -573,8 +573,7 @@ class EnumSchema(EqualByPropsMixin, NamedSchema):
raise avro.errors.InvalidName("An enum symbol must be a
valid schema name.")
if len(set(symbols)) < len(symbols):
- fail_msg = f"Duplicate symbol: {symbols}"
- raise avro.errors.AvroException(fail_msg)
+ raise avro.errors.AvroException(f"Duplicate symbol: {symbols}")
# Call parent ctor
NamedSchema.__init__(self, "enum", name, namespace, names, other_props)
@@ -584,8 +583,13 @@ class EnumSchema(EqualByPropsMixin, NamedSchema):
if doc is not None:
self.set_prop("doc", doc)
- # read-only properties
- symbols = property(lambda self: self.get_prop("symbols"))
+ @property
+ def symbols(self) -> Sequence[str]:
+ symbols = self.get_prop("symbols")
+ if isinstance(symbols, Sequence):
+ return symbols
+ raise Exception
+
doc = property(lambda self: self.get_prop("doc"))
def match(self, writer):
@@ -1108,19 +1112,21 @@ def make_avsc_object(json_data: object, names:
Optional[avro.name.Names] = None,
# JSON object (non-union)
if callable(getattr(json_data, "get", None)):
json_data = cast(Mapping, json_data)
- type = json_data.get("type")
+ type_ = json_data.get("type")
other_props = get_other_props(json_data, SCHEMA_RESERVED_PROPS)
logical_type = json_data.get("logicalType")
if logical_type:
- logical_schema = make_logical_schema(logical_type, type,
other_props or {})
+ logical_schema = make_logical_schema(logical_type, type_,
other_props or {})
if logical_schema is not None:
return cast(Schema, logical_schema)
- if type in NAMED_TYPES:
+ if type_ in NAMED_TYPES:
name = json_data.get("name")
+ if not isinstance(name, str):
+ raise avro.errors.SchemaParseException(f"Name {name} must be a
string, but it is {type(name)}.")
namespace = json_data.get("namespace", names.default_namespace)
- if type == "fixed":
+ if type_ == "fixed":
size = json_data.get("size")
if logical_type == "decimal":
precision = json_data.get("precision")
@@ -1130,8 +1136,13 @@ def make_avsc_object(json_data: object, names:
Optional[avro.name.Names] = None,
except avro.errors.IgnoredLogicalType as warning:
warnings.warn(warning)
return FixedSchema(name, namespace, size, names, other_props)
- elif type == "enum":
+ elif type_ == "enum":
symbols = json_data.get("symbols")
+ if not isinstance(symbols, Sequence):
+ raise avro.errors.SchemaParseException(f"Enum symbols must
be a sequence of strings, but it is {type(symbols)}")
+ for symbol in symbols:
+ if not isinstance(symbol, str):
+ raise avro.errors.SchemaParseException(f"Enum symbols
must be a sequence of strings, but one symbol is a {type(symbol)}")
doc = json_data.get("doc")
return EnumSchema(
name,
@@ -1142,32 +1153,31 @@ def make_avsc_object(json_data: object, names:
Optional[avro.name.Names] = None,
other_props,
validate_enum_symbols,
)
- elif type in ["record", "error"]:
+ if type_ in ["record", "error"]:
fields = json_data.get("fields")
doc = json_data.get("doc")
- return RecordSchema(name, namespace, fields, names, type, doc,
other_props)
- else:
- raise avro.errors.SchemaParseException(f"Unknown Named Type:
{type}")
+ return RecordSchema(name, namespace, fields, names, type_,
doc, other_props)
+ raise avro.errors.SchemaParseException(f"Unknown Named Type:
{type_}")
- if type in PRIMITIVE_TYPES:
- return PrimitiveSchema(type, other_props)
+ if type_ in PRIMITIVE_TYPES:
+ return PrimitiveSchema(type_, other_props)
- if type in VALID_TYPES:
- if type == "array":
+ if type_ in VALID_TYPES:
+ if type_ == "array":
items = json_data.get("items")
return ArraySchema(items, names, other_props)
- elif type == "map":
+ elif type_ == "map":
values = json_data.get("values")
return MapSchema(values, names, other_props)
- elif type == "error_union":
+ elif type_ == "error_union":
declared_errors = json_data.get("declared_errors")
return ErrorUnionSchema(declared_errors, names)
else:
- raise avro.errors.SchemaParseException(f"Unknown Valid Type:
{type}")
- elif type is None:
+ raise avro.errors.SchemaParseException(f"Unknown Valid Type:
{type_}")
+ elif type_ is None:
raise avro.errors.SchemaParseException(f'No "type" property:
{json_data}')
else:
- raise avro.errors.SchemaParseException(f"Undefined type: {type}")
+ raise avro.errors.SchemaParseException(f"Undefined type: {type_}")
# JSON array (union)
elif isinstance(json_data, list):
return UnionSchema(json_data, names)
diff --git a/lang/py/avro/test/test_bench.py b/lang/py/avro/test/test_bench.py
index e1d4416..e9edd08 100644
--- a/lang/py/avro/test/test_bench.py
+++ b/lang/py/avro/test/test_bench.py
@@ -32,7 +32,7 @@ from typing import List, Mapping, Sequence
import avro.datafile
import avro.io
import avro.schema
-import avro.utils
+from avro.utils import randbytes
TYPES = ("A", "CNAME")
SCHEMA: avro.schema.RecordSchema = avro.schema.parse(
@@ -77,7 +77,7 @@ def rand_name() -> str:
def rand_ip() -> str:
- return ".".join(map(str, avro.utils.randbytes(4)))
+ return ".".join(map(str, randbytes(4)))
def picks(n) -> Sequence[Mapping[str, str]]:
diff --git a/lang/py/avro/test/test_io.py b/lang/py/avro/test/test_io.py
index db4d043..3ef5c2b 100644
--- a/lang/py/avro/test/test_io.py
+++ b/lang/py/avro/test/test_io.py
@@ -25,10 +25,17 @@ import itertools
import json
import unittest
import warnings
+from typing import BinaryIO, Collection, Dict, List, Optional, Tuple, Union,
cast
import avro.io
import avro.schema
import avro.timezones
+from avro.utils import TypedDict
+
+
+class DefaultValueTestCaseType(TypedDict):
+ H: object
+
SCHEMAS_TO_VALIDATE = tuple(
(json.dumps(schema), datum)
@@ -212,7 +219,7 @@ LONG_RECORD_SCHEMA = avro.schema.parse(
LONG_RECORD_DATUM = {"A": 1, "B": 2, "C": 3, "D": 4, "E": 5, "F": 6, "G": 7}
-def avro_hexlify(reader):
+def avro_hexlify(reader: BinaryIO) -> bytes:
"""Return the hex value, as a string, of a binary-encoded int or long."""
b = []
current_byte = reader.read(1)
@@ -223,7 +230,7 @@ def avro_hexlify(reader):
return b" ".join(b)
-def write_datum(datum, writers_schema):
+def write_datum(datum: object, writers_schema: avro.schema.Schema) ->
Tuple[io.BytesIO, avro.io.BinaryEncoder, avro.io.DatumWriter]:
writer = io.BytesIO()
encoder = avro.io.BinaryEncoder(writer)
datum_writer = avro.io.DatumWriter(writers_schema)
@@ -231,7 +238,7 @@ def write_datum(datum, writers_schema):
return writer, encoder, datum_writer
-def read_datum(buffer, writers_schema, readers_schema=None):
+def read_datum(buffer: io.BytesIO, writers_schema: avro.schema.Schema,
readers_schema: Optional[avro.schema.Schema] = None) -> object:
reader = io.BytesIO(buffer.getvalue())
decoder = avro.io.BinaryDecoder(reader)
datum_reader = avro.io.DatumReader(writers_schema, readers_schema)
@@ -239,7 +246,7 @@ def read_datum(buffer, writers_schema, readers_schema=None):
class IoValidateTestCase(unittest.TestCase):
- def __init__(self, test_schema, test_datum):
+ def __init__(self, test_schema: str, test_datum: object) -> None:
"""Ignore the normal signature for unittest.TestCase because we are
generating
many test cases from this one class. This is safe as long as the
autoloader
ignores this class. The autoloader will ignore this class as long as
it has
@@ -251,7 +258,7 @@ class IoValidateTestCase(unittest.TestCase):
# Never hide repeated warnings when running this test case.
warnings.simplefilter("always")
- def io_valid(self):
+ def io_valid(self) -> None:
"""
In these cases, the provided data should be valid with the given
schema.
"""
@@ -263,7 +270,7 @@ class IoValidateTestCase(unittest.TestCase):
class RoundTripTestCase(unittest.TestCase):
- def __init__(self, test_schema, test_datum):
+ def __init__(self, test_schema: str, test_datum: object) -> None:
"""Ignore the normal signature for unittest.TestCase because we are
generating
many test cases from this one class. This is safe as long as the
autoloader
ignores this class. The autoloader will ignore this class as long as
it has
@@ -275,37 +282,38 @@ class RoundTripTestCase(unittest.TestCase):
# Never hide repeated warnings when running this test case.
warnings.simplefilter("always")
- def io_round_trip(self):
+ def io_round_trip(self) -> None:
"""
A datum should be the same after being encoded and then decoded.
"""
with warnings.catch_warnings(record=True) as actual_warnings:
writer, encoder, datum_writer = write_datum(self.test_datum,
self.test_schema)
round_trip_datum = read_datum(writer, self.test_schema)
- expected, round_trip, message = (
- (
+ expected: object
+ round_trip: object
+ if isinstance(round_trip_datum, decimal.Decimal):
+ expected, round_trip, message = (
str(self.test_datum),
round_trip_datum.to_eng_string(),
"Decimal datum changed value after encode and decode",
)
- if isinstance(round_trip_datum, decimal.Decimal)
- else (
- self.test_datum.astimezone(tz=avro.timezones.utc),
+ elif isinstance(round_trip_datum, datetime.datetime):
+ expected, round_trip, message = (
+ cast(datetime.datetime,
self.test_datum).astimezone(tz=avro.timezones.utc),
round_trip_datum,
"DateTime datum changed value after encode and decode",
)
- if isinstance(round_trip_datum, datetime.datetime)
- else (
+ else:
+ expected, round_trip, message = (
self.test_datum,
round_trip_datum,
"Datum changed value after encode and decode",
)
- )
self.assertEqual(expected, round_trip, message)
class BinaryEncodingTestCase(unittest.TestCase):
- def __init__(self, skip, test_type, test_datum, test_hex):
+ def __init__(self, skip: bool, test_type: str, test_datum: object,
test_hex: bytes) -> None:
"""Ignore the normal signature for unittest.TestCase because we are
generating
many test cases from this one class. This is safe as long as the
autoloader
ignores this class. The autoloader will ignore this class as long as
it has
@@ -318,7 +326,7 @@ class BinaryEncodingTestCase(unittest.TestCase):
# Never hide repeated warnings when running this test case.
warnings.simplefilter("always")
- def check_binary_encoding(self):
+ def check_binary_encoding(self) -> None:
with warnings.catch_warnings(record=True) as actual_warnings:
writer, encoder, datum_writer = write_datum(self.test_datum,
self.writers_schema)
writer.seek(0)
@@ -329,7 +337,7 @@ class BinaryEncodingTestCase(unittest.TestCase):
"Binary encoding did not match expected hex representation.",
)
- def check_skip_encoding(self):
+ def check_skip_encoding(self) -> None:
VALUE_TO_READ = 6253
with warnings.catch_warnings(record=True) as actual_warnings:
# write the value to skip and a known value
@@ -353,7 +361,7 @@ class BinaryEncodingTestCase(unittest.TestCase):
class SchemaPromotionTestCase(unittest.TestCase):
- def __init__(self, write_type, read_type):
+ def __init__(self, write_type: str, read_type: str) -> None:
"""Ignore the normal signature for unittest.TestCase because we are
generating
many test cases from this one class. This is safe as long as the
autoloader
ignores this class. The autoloader will ignore this class as long as
it has
@@ -365,7 +373,7 @@ class SchemaPromotionTestCase(unittest.TestCase):
# Never hide repeated warnings when running this test case.
warnings.simplefilter("always")
- def check_schema_promotion(self):
+ def check_schema_promotion(self) -> None:
"""Test schema promotion"""
# note that checking writers_schema.type in read_data
# allows us to handle promotion correctly
@@ -381,7 +389,7 @@ class SchemaPromotionTestCase(unittest.TestCase):
class DefaultValueTestCase(unittest.TestCase):
- def __init__(self, field_type, default):
+ def __init__(self, field_type: Collection[str], default: Union[Dict[str,
int], List[int], None, float, str]) -> None:
"""Ignore the normal signature for unittest.TestCase because we are
generating
many test cases from this one class. This is safe as long as the
autoloader
ignores this class. The autoloader will ignore this class as long as
it has
@@ -393,9 +401,10 @@ class DefaultValueTestCase(unittest.TestCase):
# Never hide repeated warnings when running this test case.
warnings.simplefilter("always")
- def check_default_value(self):
+ def check_default_value(self) -> None:
+ datum_read: DefaultValueTestCaseType
with warnings.catch_warnings(record=True) as actual_warnings:
- datum_to_read = {"H": self.default}
+ datum_to_read = cast(DefaultValueTestCaseType, {"H": self.default})
readers_schema = avro.schema.parse(
json.dumps(
{
@@ -412,12 +421,13 @@ class DefaultValueTestCase(unittest.TestCase):
)
)
writer, _, _ = write_datum(LONG_RECORD_DATUM, LONG_RECORD_SCHEMA)
- datum_read = read_datum(writer, LONG_RECORD_SCHEMA, readers_schema)
+ datum_read_ = cast(DefaultValueTestCaseType, read_datum(writer,
LONG_RECORD_SCHEMA, readers_schema))
+ datum_read = {"H": cast(bytes, datum_read_["H"]).decode()} if
isinstance(datum_read_["H"], bytes) else datum_read_
self.assertEqual(datum_to_read, datum_read)
class TestMisc(unittest.TestCase):
- def test_decimal_bytes_small_scale(self):
+ def test_decimal_bytes_small_scale(self) -> None:
"""Avro should raise an AvroTypeException when attempting to write a
decimal with a larger exponent than the schema's scale."""
datum = decimal.Decimal("3.1415")
_, _, exp = datum.as_tuple()
@@ -434,7 +444,7 @@ class TestMisc(unittest.TestCase):
)
self.assertRaises(avro.errors.AvroOutOfScaleException, write_datum,
datum, schema)
- def test_decimal_fixed_small_scale(self):
+ def test_decimal_fixed_small_scale(self) -> None:
"""Avro should raise an AvroTypeException when attempting to write a
decimal with a larger exponent than the schema's scale."""
datum = decimal.Decimal("3.1415")
_, _, exp = datum.as_tuple()
@@ -453,7 +463,7 @@ class TestMisc(unittest.TestCase):
)
self.assertRaises(avro.errors.AvroOutOfScaleException, write_datum,
datum, schema)
- def test_unknown_symbol(self):
+ def test_unknown_symbol(self) -> None:
datum_to_write = "FOO"
writers_schema = avro.schema.parse(json.dumps({"type": "enum", "name":
"Test", "symbols": ["FOO", "BAR"]}))
readers_schema = avro.schema.parse(json.dumps({"type": "enum", "name":
"Test", "symbols": ["BAR", "BAZ"]}))
@@ -464,7 +474,7 @@ class TestMisc(unittest.TestCase):
datum_reader = avro.io.DatumReader(writers_schema, readers_schema)
self.assertRaises(avro.errors.SchemaResolutionException,
datum_reader.read, decoder)
- def test_no_default_value(self):
+ def test_no_default_value(self) -> None:
writers_schema = LONG_RECORD_SCHEMA
datum_to_write = LONG_RECORD_DATUM
@@ -484,7 +494,7 @@ class TestMisc(unittest.TestCase):
datum_reader = avro.io.DatumReader(writers_schema, readers_schema)
self.assertRaises(avro.errors.SchemaResolutionException,
datum_reader.read, decoder)
- def test_projection(self):
+ def test_projection(self) -> None:
writers_schema = LONG_RECORD_SCHEMA
datum_to_write = LONG_RECORD_DATUM
@@ -506,7 +516,7 @@ class TestMisc(unittest.TestCase):
datum_read = read_datum(writer, writers_schema, readers_schema)
self.assertEqual(datum_to_read, datum_read)
- def test_field_order(self):
+ def test_field_order(self) -> None:
writers_schema = LONG_RECORD_SCHEMA
datum_to_write = LONG_RECORD_DATUM
@@ -528,7 +538,7 @@ class TestMisc(unittest.TestCase):
datum_read = read_datum(writer, writers_schema, readers_schema)
self.assertEqual(datum_to_read, datum_read)
- def test_type_exception(self):
+ def test_type_exception(self) -> None:
writers_schema = avro.schema.parse(
json.dumps(
{
@@ -545,7 +555,7 @@ class TestMisc(unittest.TestCase):
self.assertRaises(avro.errors.AvroTypeException, write_datum,
datum_to_write, writers_schema)
-def load_tests(loader, default_tests, pattern):
+def load_tests(loader: unittest.TestLoader, default_tests: None, pattern:
None) -> unittest.TestSuite:
"""Generate test cases across many test schema."""
suite = unittest.TestSuite()
suite.addTests(loader.loadTestsFromTestCase(TestMisc))
diff --git a/lang/py/avro/utils.py b/lang/py/avro/utils.py
index 76cc8c7..76d8f6e 100644
--- a/lang/py/avro/utils.py
+++ b/lang/py/avro/utils.py
@@ -22,6 +22,12 @@ Arbitrary utilities and polyfills.
"""
import random
+import sys
+
+if sys.version_info >= (3, 8):
+ from typing import TypedDict
+else:
+ from typing_extensions import TypedDict
def _randbytes(n: int) -> bytes:
