datetime logical types support
MODIFY
fastavro/reader.py
fastavro/reader.py
@@ -10,6 +10,13 @@ import json
from struct import unpack, error as StructError
from zlib import decompress
+try:
+ from UserDict import UserDict
+except ImportError:
+ from collections import UserDict
+
+import datetime
+
try:
from fastavro._six import MemoryIO, xrange, btou, utob, iteritems, is_str
from fastavro._schema import (
@@ -134,6 +141,23 @@ def read_boolean(fo, writer_schema=None, reader_schema=None):
return unpack('B', fo.read(1))[0] != 0
+def read_timestamp_millis(fo, writer_schema=None, reader_schema=None):
+ res = read_long(fo, writer_schema, reader_schema)
+ return datetime.datetime.fromtimestamp(res / 1000).replace(
+ microsecond=res % 1000)
+
+
+def read_timestamp_micros(fo, writer_schema=None, reader_schema=None):
+ res = read_long(fo, writer_schema, reader_schema)
+ return datetime.datetime.fromtimestamp(res/1000000).replace(
+ microsecond=res % 1000000)
+
+
+def read_date(fo, writer_schema=None, reader_schema=None):
+ res = read_long(fo, writer_schema, reader_schema)
+ return datetime.date.fromordinal(res)
+
+
def read_long(fo, writer_schema=None, reader_schema=None):
"""int and long values are written using variable-length, zig-zag
coding."""
@@ -350,7 +374,33 @@ def read_record(fo, writer_schema, reader_schema=None):
return record
-READERS = {
+LOGICALREADERS = {
+ 'long-timestamp-millis': read_timestamp_millis,
+ 'long-timestamp-micros': read_timestamp_micros,
+ 'int-date': read_date
+}
+
+
+class Readers(UserDict):
+ def __getitem__(self, key):
+ rt = extract_record_type(key)
+ if rt == 'record':
+ return self.data[rt]
+ try:
+ return self.data.__getitem__(key)
+ except:
+ pass
+ if isinstance(key, dict):
+ if "logicalType" not in key:
+ return self.data[rt]
+ return LOGICALREADERS[key['type'] + "-" + key['logicalType']]
+ if isinstance(key, list):
+ return self.data['union']
+
+
+READERS = Readers()
+
+READERS.update({
'null': read_null,
'boolean': read_boolean,
'string': read_utf8,
@@ -368,7 +418,7 @@ READERS = {
'record': read_record,
'error': read_record,
'request': read_record,
-}
+})
def read_data(fo, writer_schema, reader_schema=None):
@@ -378,7 +428,8 @@ def read_data(fo, writer_schema, reader_schema=None):
if reader_schema and record_type in AVRO_TYPES:
match_schemas(writer_schema, reader_schema)
try:
- return READERS[record_type](fo, writer_schema, reader_schema)
+ fn = READERS[writer_schema]
+ return fn(fo, writer_schema, reader_schema)
except StructError:
raise EOFError('cannot read %s from %s' % (record_type, fo))

MODIFY
fastavro/writer.py
fastavro/writer.py
# http://svn.apache.org/viewvc/avro/trunk/lang/py/src/avro/ which is under
# Apache 2.0 license (http://www.apache.org/licenses/LICENSE-2.0)
+try:
+ from UserDict import UserDict
+except ImportError:
+ from collections import UserDict
+
try:
from fastavro._six import utob, MemoryIO, long, is_str, iteritems
from fastavro._reader import HEADER_SCHEMA, SYNC_SIZE, MAGIC
@@ -22,6 +27,7 @@ try:
except ImportError:
import json
+import time
from binascii import crc32
from collections import Iterable, Mapping
from os import urandom, SEEK_SET
@@ -42,6 +48,23 @@ def write_boolean(fo, datum, schema=None):
fo.write(pack('B', 1 if datum else 0))
+def write_timestamp_millis(fo, datum, schema=None):
+ t = int(time.mktime(datum.timetuple())) * 1000 + int(
+ datum.microsecond / 1000)
+ write_int(fo, t, schema)
+ # write_int(fo, int(datum.timestamp() * 1000), schema)
+
+
+def write_timestamp_micros(fo, datum, schema=None):
+ t = int(time.mktime(datum.timetuple())) * 1000000 + datum.microsecond
+ write_int(fo, t, schema)
+ # write_int(fo, int(datum.timestamp() * 1000000), schema)
+
+
+def write_date(fo, datum, schema=None):
+ write_int(fo, datum.toordinal(), schema)
+
+
def write_int(fo, datum, schema=None):
"""int and long values are written using variable-length, zig-zag coding.
"""
@@ -264,7 +287,33 @@ def write_record(fo, datum, schema):
name, field.get('default')), field['type'])
-WRITERS = {
+LOGICALWRITERS = {
+ 'long-timestamp-millis': write_timestamp_millis,
+ 'long-timestamp-micros': write_timestamp_micros,
+ 'int-date': write_date
+}
+
+
+class Writers(UserDict):
+ def __getitem__(self, key):
+ rt = extract_record_type(key)
+ if rt == 'record':
+ return self.data[rt]
+ try:
+ return self.data.__getitem__(key)
+ except:
+ pass
+ if isinstance(key, dict):
+ if "logicalType" not in key:
+ return self.data[rt]
+ return LOGICALWRITERS[key['type'] + "-" + key['logicalType']]
+ if isinstance(key, list):
+ return self.data['union']
+
+
+WRITERS = Writers()
+
+WRITERS.update({
'null': write_null,
'boolean': write_boolean,
'string': write_utf8,
@@ -281,7 +330,7 @@ WRITERS = {
'error_union': write_union,
'record': write_record,
'error': write_record,
-}
+})
_base_types = [
'boolean',
@@ -309,7 +358,8 @@ def write_data(fo, datum, schema):
schema: dict
Schemda to use
"""
- return WRITERS[extract_record_type(schema)](fo, datum, schema)
+ fn = WRITERS[schema]
+ return fn(fo, datum, schema)
def write_header(fo, metadata, sync_marker):

ADD
tests/test_logical_types.py
+import datetime
+from io import BytesIO
+
+import fastavro
+
+schema = {
+ "fields": [
+ {
+ "name": "date",
+ "type": {'type': 'int', 'logicalType': 'date'}
+ },
+ {
+ "name": "timestamp-millis",
+ "type": {'type': 'long', 'logicalType': 'timestamp-millis'}
+ },
+ {
+ "name": "timestamp-micros",
+ "type": {'type': 'long', 'logicalType': 'timestamp-micros'}
+ },
+ ],
+ "namespace": "namespace",
+ "name": "name",
+ "type": "record"
+}
+
+
+def serialize(schema, data):
+ bytes_writer = BytesIO()
+ fastavro.schemaless_writer(bytes_writer, schema, data)
+ return bytes_writer.getvalue()
+
+
+def deserialize(schema, binary):
+ bytes_writer = BytesIO()
+ bytes_writer.write(binary)
+ bytes_writer.seek(0)
+
+ res = fastavro.schemaless_reader(bytes_writer, schema)
+ return res
+
+
+def test_logical_types():
+ data1 = {
+ 'date': datetime.date.today(),
+ 'timestamp-millis': datetime.datetime.now(),
+ 'timestamp-micros': datetime.datetime.now(),
+
+ }
+ binary = serialize(schema, data1)
+ data2 = deserialize(schema, binary)
+ assert (data1['date'] == data2['date'])
+ assert (data1['timestamp-micros'] == data2['timestamp-micros'])
+ assert (int(data1['timestamp-millis'].microsecond / 1000) == int(data2[
+ 'timestamp-millis'].microsecond))
+
+
+test_logical_types()

datetime logical types support