Unknown Logical Types
Fixing logical types lookup from Key lookup to .get(key) with conditions to only prepare if logicalType exists.
MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -54,7 +54,6 @@ AVRO_TYPES = set([
'error_union'
])
-
ctypedef int int32
ctypedef unsigned int uint32
ctypedef unsigned long long ulong64
@@ -79,7 +78,6 @@ cpdef match_types(writer_type, reader_type):
return True
return False
-
cpdef match_schemas(w_schema, r_schema):
error_msg = 'Schema mismatch: %s is not %s' % (w_schema, r_schema)
if isinstance(w_schema, list):
@@ -115,12 +113,10 @@ cpdef match_schemas(w_schema, r_schema):
return True
raise SchemaResolutionError(error_msg)
-
cpdef inline read_null(ReaderBase fo, writer_schema=None, reader_schema=None):
"""null is written as zero bytes."""
return None
-
cpdef inline read_boolean(ReaderBase fo, writer_schema=None, reader_schema=None):
"""A boolean is written as a single byte whose value is either 0 (false) or
1 (true).
@@ -135,27 +131,21 @@ cpdef inline read_boolean(ReaderBase fo, writer_schema=None, reader_schema=None)
else:
raise ReadError
-
cpdef parse_timestamp(data, resolution):
return datetime.datetime.fromtimestamp(data / resolution, tz=utc)
-
cpdef read_timestamp_millis(data, writer_schema=None, reader_schema=None):
return parse_timestamp(data, float(MLS_PER_SECOND))
-
cpdef read_timestamp_micros(data, writer_schema=None, reader_schema=None):
return parse_timestamp(data, float(MCS_PER_SECOND))
-
cpdef read_date(data, writer_schema=None, reader_schema=None):
return datetime.date.fromordinal(data + DAYS_SHIFT)
-
cpdef read_uuid(data, writer_schema=None, reader_schema=None):
return UUID(data)
-
cpdef read_time_millis(data, writer_schema=None, reader_schema=None):
h = int(data / MLS_PER_HOUR)
m = int(data / MLS_PER_MINUTE) % 60
@@ -163,7 +153,6 @@ cpdef read_time_millis(data, writer_schema=None, reader_schema=None):
mls = int(data % MLS_PER_SECOND) * 1000
return datetime.time(h, m, s, mls)
-
cpdef read_time_micros(data, writer_schema=None, reader_schema=None):
h = int(data / MCS_PER_HOUR)
m = int(data / MCS_PER_MINUTE) % 60
@@ -171,17 +160,14 @@ cpdef read_time_micros(data, writer_schema=None, reader_schema=None):
mcs = data % MCS_PER_SECOND
return datetime.time(h, m, s, mcs)
-
cpdef read_bytes_decimal(data, writer_schema=None, reader_schema=None):
size = len(data)
return _read_decimal(data, size, writer_schema)
-
cpdef read_fixed_decimal(data, writer_schema=None, reader_schema=None):
size = writer_schema['size']
return _read_decimal(data, size, writer_schema)
-
cpdef _read_decimal(data, size, writer_schema):
"""
based on https://github.com/apache/avro/pull/82/
@@ -212,7 +198,6 @@ cpdef _read_decimal(data, size, writer_schema):
scaled_datum = Decimal(unscaled_datum).scaleb(-scale)
return scaled_datum
-
cpdef long64 read_long(ReaderBase fo,
writer_schema=None,
reader_schema=None) except? -1:
@@ -239,12 +224,10 @@ cpdef long64 read_long(ReaderBase fo,
return (n >> 1) ^ -(n & 1)
-
cdef union float_uint32:
float f
uint32 n
-
cpdef read_float(ReaderBase fo, writer_schema=None, reader_schema=None):
"""A float is written as 4 bytes.
@@ -265,12 +248,10 @@ cpdef read_float(ReaderBase fo, writer_schema=None, reader_schema=None):
else:
raise ReadError
-
cdef union double_ulong64:
double d
ulong64 n
-
cpdef read_double(ReaderBase fo, writer_schema=None, reader_schema=None):
"""A double is written as 8 bytes.
@@ -295,26 +276,22 @@ cpdef read_double(ReaderBase fo, writer_schema=None, reader_schema=None):
else:
raise ReadError
-
cpdef read_bytes(ReaderBase fo, writer_schema=None, reader_schema=None):
"""Bytes are encoded as a long followed by that many bytes of data."""
cdef long64 size = read_long(fo)
return fo.read(<long> size)
-
cpdef unicode read_utf8(ReaderBase fo, writer_schema=None, reader_schema=None):
"""A string is encoded as a long followed by that many bytes of UTF-8
encoded character data.
"""
return btou(read_bytes(fo), 'utf-8')
-
cpdef read_fixed(ReaderBase fo, writer_schema, reader_schema=None):
"""Fixed instances are encoded using the number of bytes declared in the
schema."""
return fo.read(writer_schema['size'])
-
cpdef read_enum(ReaderBase fo, writer_schema, reader_schema=None):
"""An enum is encoded by a int, representing the zero-based position of the
symbol in the schema.
@@ -327,7 +304,6 @@ cpdef read_enum(ReaderBase fo, writer_schema, reader_schema=None):
raise SchemaResolutionError(msg)
return symbol
-
cpdef read_array(ReaderBase fo, writer_schema, reader_schema=None):
"""Arrays are encoded as a series of blocks.
@@ -365,7 +341,6 @@ cpdef read_array(ReaderBase fo, writer_schema, reader_schema=None):
return read_items
-
cpdef read_map(ReaderBase fo, writer_schema, reader_schema=None):
"""Maps are encoded as a series of blocks.
@@ -404,7 +379,6 @@ cpdef read_map(ReaderBase fo, writer_schema, reader_schema=None):
return read_items
-
cpdef read_union(ReaderBase fo, writer_schema, reader_schema=None):
"""A union is encoded by first writing a long value indicating the
zero-based position within the union of the schema of its value.
@@ -428,7 +402,6 @@ cpdef read_union(ReaderBase fo, writer_schema, reader_schema=None):
else:
return _read_data(fo, writer_schema[index])
-
cpdef read_record(ReaderBase fo, writer_schema, reader_schema=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
@@ -487,7 +460,6 @@ cpdef read_record(ReaderBase fo, writer_schema, reader_schema=None):
return record
-
LOGICAL_READERS = {
'long-timestamp-millis': read_timestamp_millis,
'long-timestamp-micros': read_timestamp_micros,
@@ -519,11 +491,9 @@ READERS = {
'request': read_record,
}
-
cpdef read_data(fo, writer_schema, reader_schema=None):
return _read_data(FileObjectReader(fo), writer_schema, reader_schema)
-
cpdef _read_data(ReaderBase fo, writer_schema, reader_schema=None):
"""Read data from file object according to schema."""
@@ -535,25 +505,23 @@ cpdef _read_data(ReaderBase fo, writer_schema, reader_schema=None):
try:
data = READERS[record_type](fo, writer_schema, reader_schema)
if 'logicalType' in writer_schema:
- fn = LOGICAL_READERS[logical_type]
+ fn = LOGICAL_READERS.get(logical_type)
+ if fn:
return fn(data, writer_schema, reader_schema)
return data
except ReadError:
raise EOFError('cannot read %s from %s' % (record_type, fo))
-
cpdef skip_sync(ReaderBase fo, sync_marker):
"""Skip an expected sync marker, complaining if it doesn't match"""
if fo.read(SYNC_SIZE) != sync_marker:
raise ValueError('expected sync marker not found')
-
cdef class ReaderBase(object):
cpdef bytes read(self, int32 size):
raise NotImplemented
-
cdef class MemoryReader(ReaderBase):
cdef bytes data
cdef ulong64 position
@@ -568,7 +536,6 @@ cdef class MemoryReader(ReaderBase):
self.position += size
return result
-
cdef class FileObjectReader(ReaderBase):
cdef object fo
@@ -578,14 +545,12 @@ cdef class FileObjectReader(ReaderBase):
cpdef bytes read(self, int32 size):
return self.fo.read(size)
-
cpdef MemoryReader null_read_block(fo):
"""Read block in "null" codec."""
length = read_long(fo, None)
data = fo.read(length)
return MemoryReader(data)
-
cpdef MemoryReader deflate_read_block(fo):
"""Read block in "deflate" codec."""
data = read_bytes(fo, None)
@@ -593,7 +558,6 @@ cpdef MemoryReader deflate_read_block(fo):
# zlib headers) decompression. See zlib.h.
return MemoryReader(decompress(data, -15))
-
BLOCK_READERS = {
'null': null_read_block,
'deflate': deflate_read_block
@@ -686,12 +650,10 @@ class reader:
# Deprecated
iter_avro = reader
-
cpdef schemaless_reader(fo, schema):
acquaint_schema(schema)
return read_data(fo, schema)
-
cpdef is_avro(path_or_buffer):
if is_str(path_or_buffer):
fp = open(path_or_buffer, 'rb')

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -464,9 +464,9 @@ def read_data(fo, writer_schema, reader_schema=None):
try:
data = READERS[record_type](fo, writer_schema, reader_schema)
if 'logicalType' in writer_schema:
- fn = LOGICAL_READERS[logical_type]
+ fn = LOGICAL_READERS.get(logical_type)
+ if fn:
return fn(data, writer_schema, reader_schema)
-
return data
except StructError:
raise EOFError('cannot read %s from %s' % (record_type, fo))

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
@@ -626,7 +626,8 @@ cpdef write_data(bytearray fo, datum, schema):
if isinstance(schema, dict):
logical_type = extract_logical_type(schema)
if logical_type:
- prepare = LOGICAL_WRITERS[logical_type]
+ prepare = LOGICAL_WRITERS.get(logical_type)
+ if prepare:
datum = prepare(datum, schema)
record_type = extract_record_type(schema)

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -11,6 +11,7 @@ import json
import datetime
import decimal
import time
+import uuid
from binascii import crc32
from collections import Iterable, Mapping
from os import urandom, SEEK_SET
@@ -46,6 +47,8 @@ def write_boolean(fo, datum, schema=None):
def prepare_timestamp_millis(data, schema):
+ """Converts datetime.datetime object to int timestamp with milliseconds
+ """
if isinstance(data, datetime.datetime):
if data.tzinfo is not None:
delta = (data - epoch)
@@ -58,6 +61,7 @@ def prepare_timestamp_millis(data, schema):
def prepare_timestamp_micros(data, schema):
+ """Converts datetime.datetime to int timestamp with microseconds"""
if isinstance(data, datetime.datetime):
if data.tzinfo is not None:
delta = (data - epoch)
@@ -70,6 +74,7 @@ def prepare_timestamp_micros(data, schema):
def prepare_date(data, schema):
+ """Converts datetime.date to int timestamp"""
if isinstance(data, datetime.date):
return data.toordinal() - DAYS_SHIFT
else:
@@ -77,10 +82,17 @@ def prepare_date(data, schema):
def prepare_uuid(data, schema):
+ """Converts uuid.UUID to
+ string formatted UUID xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
+ """
+ if isinstance(data, uuid.UUID):
return str(data)
+ else:
+ return data
def prepare_time_millis(data, schema):
+ """Convert datetime.time to int timestamp with milliseconds"""
if isinstance(data, datetime.time):
return int(
data.hour * MLS_PER_HOUR + data.minute * MLS_PER_MINUTE +
@@ -90,6 +102,7 @@ def prepare_time_millis(data, schema):
def prepare_time_micros(data, schema):
+ """Convert datetime.time to int timestamp with microseconds"""
if isinstance(data, datetime.time):
return long(data.hour * MCS_PER_HOUR + data.minute * MCS_PER_MINUTE +
data.second * MCS_PER_SECOND + data.microsecond)
@@ -98,6 +111,7 @@ def prepare_time_micros(data, schema):
def prepare_bytes_decimal(data, schema):
+ """Convert decimal.Decimal to bytes"""
if not isinstance(data, decimal.Decimal):
return data
scale = schema['scale']
@@ -138,6 +152,7 @@ def prepare_bytes_decimal(data, schema):
def prepare_fixed_decimal(data, schema):
+ """Converts decimal.Decimal to fixed length bytes array"""
if not isinstance(data, decimal.Decimal):
return data
scale = schema['scale']
@@ -488,9 +503,9 @@ def write_data(fo, datum, schema):
fn = WRITERS[record_type]
if logical_type:
- prepare = LOGICAL_WRITERS[logical_type]
- data = prepare(datum, schema)
- return fn(fo, data, schema)
+ prepare = LOGICAL_WRITERS.get(logical_type)
+ if prepare:
+ datum = prepare(datum, schema)
return fn(fo, datum, schema)
@@ -527,7 +542,6 @@ BLOCK_WRITERS = {
'deflate': deflate_write_block
}
-
try:
import snappy

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
@@ -11,7 +11,6 @@ import os
from .conftest import assert_naive_datetime_equal_to_tz_datetime
-
schema = {
"fields": [
{
@@ -283,3 +282,58 @@ def test_clean_json_dict():
}
_clean_json_record(values)
assert values == str_values
+
+
+def test_unknown_logical_type():
+ unknown_type_schema = {
+ "type": "record",
+ "name": "t_customers",
+ "fields": [{
+ "name": "name",
+ "type": {
+ "type": "string",
+ # "logicalType": "varchar",
+ "maxLength": 200
+ },
+ }, {
+ "name": "address",
+ "type": {
+ "type": "record",
+ "name": "t_address",
+ "fields": [{
+ "name": "street",
+ "type": {
+ "type": "string",
+ # "logicalType": "varchar",
+ "maxLength": 240
+ }
+ }, {
+ "name": "city",
+ "type": {
+ "type": "string",
+ # "logicalType": "varchar",
+ "maxLength": 80
+ }
+ }, {
+ "name": "zip",
+ "type": {
+ "type": "string",
+ # "logicalType": "varchar",
+ "maxLength": 18
+ }
+ }]
+ }
+ }]
+ }
+
+ data1 = {
+ 'name': 'foo',
+ 'address': {
+ 'street': '123 street',
+ 'city': 'city',
+ 'zip': '00000'
+ }
+ }
+ converted = serialize(unknown_type_schema, data1)
+ data2 = deserialize(unknown_type_schema, converted)
+ assert (data1 == data2)

Unknown Logical Types

#178 https://github.com/tebeka/fastavro/issues/178

Fixing logical types lookup from Key lookup to .get(key) with conditions to only prepare if logicalType exists.