allow for schema migration in the schemaless reader
MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -687,9 +687,11 @@ class reader:
iter_avro = reader
-cpdef schemaless_reader(fo, schema):
- acquaint_schema(schema)
- return read_data(fo, schema)
+cpdef schemaless_reader(fo, writer_schema, reader_schema=None):
+ acquaint_schema(writer_schema)
+ if reader_schema:
+ populate_schema_defs(reader_schema)
+ return read_data(fo, writer_schema, reader_schema)
cpdef is_avro(path_or_buffer):

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -602,15 +602,18 @@ class reader:
iter_avro = reader
-def schemaless_reader(fo, schema):
+def schemaless_reader(fo, writer_schema, reader_schema=None):
"""Reads a single record writen using the schemaless_writer
Parameters
----------
fo: file-like
Input stream
- schema: dict
- Reader schema
+ writer_schema: dict
+ Schema used when calling schemaless_writer
+ reader_schema: dict, optional
+ If the schema has changed since being written then the new schema can
+ be given to allow for schema migration
@@ -621,8 +624,10 @@ def schemaless_reader(fo, schema):
Note: The ``schemaless_reader`` can only read a single record.
"""
- acquaint_schema(schema)
- return read_data(fo, schema)
+ acquaint_schema(writer_schema)
+ if reader_schema:
+ populate_schema_defs(reader_schema)
+ return read_data(fo, writer_schema, reader_schema)
def is_avro(path_or_buffer):

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -254,24 +254,6 @@ def test_missing_schema():
fastavro.schema.load_schema(schema_path)
-def test_schemaless_writer_and_reader():
- schema = {
- "type": "record",
- "name": "Test",
- "namespace": "test",
- "fields": [{
- "name": "field",
- "type": {"type": "string"}
- }]
- }
- record = {"field": "test"}
- new_file = MemoryIO()
- fastavro.schemaless_writer(new_file, schema, record)
- new_file.seek(0)
- new_record = fastavro.schemaless_reader(new_file, schema)
- assert record == new_record
-
-
def test_default_values():
schema = {
"type": "record",
@@ -306,29 +288,6 @@ def test_nullable_values():
'nullable_field': 'no_null', 'field': 'val'}]
-def test_boolean_roundtrip():
- schema = {
- "type": "record",
- "fields": [{
- "name": "field",
- "type": "boolean"
- }]
- }
- record = {"field": True}
- new_file = MemoryIO()
- fastavro.schemaless_writer(new_file, schema, record)
- new_file.seek(0)
- new_record = fastavro.schemaless_reader(new_file, schema)
- assert record == new_record
-
- record = {"field": False}
- new_file = MemoryIO()
- fastavro.schemaless_writer(new_file, schema, record)
- new_file.seek(0)
- new_record = fastavro.schemaless_reader(new_file, schema)
- assert record == new_record
-
-
def test_metadata():
schema = {
"type": "record",

ADD
tests/test_schemaless.py
+import fastavro
+
+from fastavro.six import MemoryIO
+
+import pytest
+
+pytestmark = pytest.mark.usefixtures("clean_readers_writers_and_schemas")
+
+
+def test_schemaless_writer_and_reader():
+ schema = {
+ "type": "record",
+ "name": "Test",
+ "namespace": "test",
+ "fields": [{
+ "name": "field",
+ "type": {"type": "string"}
+ }]
+ }
+ record = {"field": "test"}
+ new_file = MemoryIO()
+ fastavro.schemaless_writer(new_file, schema, record)
+ new_file.seek(0)
+ new_record = fastavro.schemaless_reader(new_file, schema)
+ assert record == new_record
+
+
+def test_boolean_roundtrip():
+ schema = {
+ "type": "record",
+ "fields": [{
+ "name": "field",
+ "type": "boolean"
+ }]
+ }
+ record = {"field": True}
+ new_file = MemoryIO()
+ fastavro.schemaless_writer(new_file, schema, record)
+ new_file.seek(0)
+ new_record = fastavro.schemaless_reader(new_file, schema)
+ assert record == new_record
+
+ record = {"field": False}
+ new_file = MemoryIO()
+ fastavro.schemaless_writer(new_file, schema, record)
+ new_file.seek(0)
+ new_record = fastavro.schemaless_reader(new_file, schema)
+ assert record == new_record
+
+
+def test_default_values_in_reader():
+ writer_schema = {
+ 'name': 'name1',
+ 'type': 'record',
+ 'namespace': 'namespace1',
+ 'fields': [{
+ 'doc': 'test',
+ 'type': 'int',
+ 'name': 'good_field'
+ }],
+ 'doc': 'test'
+ }
+
+ reader_schema = {
+ 'name': 'name1',
+ 'doc': 'test',
+ 'namespace': 'namespace1',
+ 'fields': [{
+ 'name': 'good_field',
+ 'doc': 'test',
+ 'type': 'int'
+ }, {
+ 'name': 'good_compatible_field',
+ 'doc': 'test',
+ 'default': 1,
+ 'type': 'int'
+ }],
+ 'type': 'record'
+ }
+
+ record = {'good_field': 1}
+ new_file = MemoryIO()
+ fastavro.schemaless_writer(new_file, writer_schema, record)
+ new_file.seek(0)
+ new_record = fastavro.schemaless_reader(
+ new_file,
+ writer_schema,
+ reader_schema,
+ )
+ assert new_record == {'good_field': 1, 'good_compatible_field': 1}

allow for schema migration in the schemaless reader