decimal logical type
MODIFY
fastavro/reader.py
fastavro/reader.py
import json
from struct import unpack, error as StructError
from zlib import decompress
-
+from os import SEEK_SET
import datetime
+from decimal import getcontext, Decimal
+
+import struct
try:
- from fastavro._six import MemoryIO, xrange, btou, utob, iteritems, is_str
+ from fastavro._six import MemoryIO, xrange, btou, utob, iteritems,\
+ is_str, PY3
from fastavro._schema import (
extract_record_type, acquaint_schema, populate_schema_defs,
extract_logical_type
)
except ImportError:
- from fastavro.six import MemoryIO, xrange, btou, utob, iteritems, is_str
+ from fastavro.six import MemoryIO, xrange, btou, utob, iteritems, \
+ is_str, PY3
from fastavro.schema import (
extract_record_type, acquaint_schema, populate_schema_defs,
extract_logical_type
@@ -152,6 +157,79 @@ def read_date(data, writer_schema=None, reader_schema=None):
return datetime.date.fromordinal(data)
+def read_bytes_decimal(data, writer_schema=None, reader_schema=None):
+ scale = writer_schema['scale']
+ precision = writer_schema['precision']
+
+ tmp = MemoryIO()
+ tmp.write(data)
+
+ tmp.seek(0, SEEK_SET)
+ size = read_long(tmp)
+
+ tmp.seek(0, SEEK_SET)
+
+ datum = read_bytes(tmp)
+
+ if PY3:
+ unscaled_datum = read_decimal_from_fixed(datum, precision, scale, size)
+ else:
+ unscaled_datum = read_decimal_from_fixed2(datum,
+ precision, scale, size)
+
+ original_prec = getcontext().prec
+ getcontext().prec = precision
+ scaled_datum = Decimal(unscaled_datum).scaleb(-scale)
+ getcontext().prec = original_prec
+ return scaled_datum
+
+
+def read_decimal_from_fixed(datum, precision, scale, size):
+ """
+ Decimal is encoded as fixed. Fixed instances are encoded using the
+ number of bytes declared in the schema.
+ """
+ unscaled_datum = 0
+ msb = datum[0]
+ leftmost_bit = (msb >> 7) & 1
+ if leftmost_bit == 1:
+ modified_first_byte = (datum[0]) ^ (1 << 7)
+ datum = (modified_first_byte).to_bytes(1, 'big') + datum[1:]
+ for offset in range(size):
+ unscaled_datum <<= 8
+ unscaled_datum += datum[offset]
+ unscaled_datum += pow(-2, (size * 8) - 1)
+ else:
+ for offset in range(size):
+ unscaled_datum <<= 8
+ unscaled_datum += datum[offset]
+
+ return unscaled_datum
+
+
+def read_decimal_from_fixed2(datum, precision, scale, size):
+ """
+ Decimal is encoded as fixed. Fixed instances are encoded using the
+ number of bytes declared in the schema.
+ """
+ unscaled_datum = 0
+ msb = struct.unpack('!b', datum[0])[0]
+ leftmost_bit = (msb >> 7) & 1
+ if leftmost_bit == 1:
+ modified_first_byte = ord(datum[0]) ^ (1 << 7)
+ datum = chr(modified_first_byte) + datum[1:]
+ for offset in range(size):
+ unscaled_datum <<= 8
+ unscaled_datum += ord(datum[offset])
+ unscaled_datum += pow(-2, (size*8) - 1)
+ else:
+ for offset in range(size):
+ unscaled_datum <<= 8
+ unscaled_datum += ord(datum[offset])
+
+ return unscaled_datum
+
+
def read_long(fo, writer_schema=None, reader_schema=None):
"""int and long values are written using variable-length, zig-zag
coding."""
@@ -371,7 +449,8 @@ def read_record(fo, writer_schema, reader_schema=None):
LOGICAL_READERS = {
'long-timestamp-millis': read_timestamp_millis,
'long-timestamp-micros': read_timestamp_micros,
- 'int-date': read_date
+ 'int-date': read_date,
+ 'bytes-decimal': read_bytes_decimal
}
READERS = {

MODIFY
fastavro/six.py
fastavro/six.py
@@ -10,10 +10,11 @@ from sys import stdout
_encoding = 'UTF-8'
+PY3 = False
if sys.version_info >= (3, 0):
from io import BytesIO as MemoryIO
xrange = range
-
+ PY3 = True
def py3_btou(n, encoding=_encoding):
return n.decode(encoding)

MODIFY
fastavro/writer.py
fastavro/writer.py
# Apache 2.0 license (http://www.apache.org/licenses/LICENSE-2.0)
try:
- from fastavro._six import utob, MemoryIO, long, is_str, iteritems
+ from fastavro._six import utob, MemoryIO, long, is_str, iteritems, PY3
from fastavro._reader import HEADER_SCHEMA, SYNC_SIZE, MAGIC
from fastavro._schema import extract_named_schemas_into_repo,\
extract_record_type, extract_logical_type
except ImportError:
- from fastavro.six import utob, MemoryIO, long, is_str, iteritems
+ from fastavro.six import utob, MemoryIO, long, is_str, iteritems, PY3
from fastavro.reader import HEADER_SCHEMA, SYNC_SIZE, MAGIC
from fastavro.schema import extract_named_schemas_into_repo,\
extract_record_type, extract_logical_type
-import datetime
+
try:
import simplejson as json
except ImportError:
@@ -61,6 +61,46 @@ def write_date(data, schema):
return data.toordinal()
+def write_bytes_decimal(data, schema):
+ scale = schema['scale']
+ # pprecision = schema['precision']
+
+ sign, digits, exp = data.as_tuple()
+
+ if -exp > scale:
+ raise AssertionError(
+ 'Scale provided in schema does not match the decimal')
+ delta = exp + scale
+ if delta > 0:
+ digits = digits + (0,) * delta
+
+ unscaled_datum = 0
+ for digit in digits:
+ unscaled_datum = (unscaled_datum * 10) + digit
+
+ bits_req = unscaled_datum.bit_length() + 1
+ if sign:
+ unscaled_datum = (1 << bits_req) - unscaled_datum
+
+ bytes_req = bits_req // 8
+ padding_bits = ~((1 << bits_req) - 1) if sign else 0
+ packed_bits = padding_bits | unscaled_datum
+
+ bytes_req += 1 if (bytes_req << 3) < bits_req else 0
+
+ tmp = MemoryIO()
+
+ write_long(tmp, bytes_req)
+ for index in range(bytes_req - 1, -1, -1):
+ bits_to_write = packed_bits >> (8 * index)
+ if PY3:
+ tmp.write(bytes([bits_to_write & 0xff]))
+ else:
+ tmp.write(chr(bits_to_write & 0xff))
+
+ return tmp.getvalue()
+
+
def write_int(fo, datum, schema=None):
"""int and long values are written using variable-length, zig-zag coding.
"""
@@ -286,7 +326,8 @@ def write_record(fo, datum, schema):
LOGICAL_WRITERS = {
'long-timestamp-millis': write_timestamp_millis,
'long-timestamp-micros': write_timestamp_micros,
- 'int-date': write_date
+ 'int-date': write_date,
+ 'bytes-decimal': write_bytes_decimal
}
WRITERS = {
@@ -342,7 +383,8 @@ def write_data(fo, datum, schema):
if logical_type:
prepare = LOGICAL_WRITERS[logical_type]
- return fn(fo, prepare(datum, schema), schema)
+ data = prepare(datum, schema)
+ return fn(fo, data, schema)
return fn(fo, datum, schema)

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
import datetime
+from decimal import Decimal
from io import BytesIO
import fastavro
+from nose.tools import raises
schema = {
"fields": [
@@ -76,4 +78,24 @@ def test_null():
assert (data2['date'] is None)
-test_null()
+schema_top = {
+ "name": "n",
+ "namespace": "namespace",
+ "type": "bytes",
+ "logicalType": "decimal",
+ "precision": 15,
+ "scale": 3,
+}
+
+
+def test_top():
+ data1 = Decimal("123.456")
+ binary = serialize(schema_top, data1)
+ data2 = deserialize(schema_top, binary)
+ assert (data1 == data2)
+
+
+@raises(AssertionError)
+def test_scale():
+ data1 = Decimal("123.456678") # does not fit scale
+ serialize(schema_top, data1)

decimal logical type