Skip to content

Instantly share code, notes, and snippets.

@kkirsanov
Created November 23, 2017 15:26
Show Gist options
  • Save kkirsanov/d64a3c2899675e8aa35870c608fe7e12 to your computer and use it in GitHub Desktop.
Save kkirsanov/d64a3c2899675e8aa35870c608fe7e12 to your computer and use it in GitHub Desktop.
Disable logical reader in fastavro
import datetime
from io import BytesIO
from pprint import pprint
from uuid import uuid4
import fastavro
fastavro._reader.LOGICAL_READERS['long-timestamp-millis'] = lambda d, w, r: d
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'}
},
{
"name": "uuid",
"type": {'type': 'string', 'logicalType': 'uuid'}
},
{
"name": "time-millis",
"type": {'type': 'int', 'logicalType': 'time-millis'}
},
{
"name": "time-micros",
"type": {'type': 'long', 'logicalType': 'time-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
data1 = {
'date': datetime.date.today(),
'timestamp-millis': datetime.datetime.now(),
'timestamp-micros': datetime.datetime.now(),
'uuid': uuid4(),
'time-millis': datetime.datetime.now().time(),
'time-micros': datetime.datetime.now().time(),
}
binary = serialize(schema, data1)
data2 = deserialize(schema, binary)
pprint(data2)
@NoAnyLove
Copy link

I think line 8 should be,

fastavro._read.LOGICAL_READERS['long-timestamp-millis'] = lambda d, w, r: d

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment