ensure acquaint_schema acquints to both READERS and WRITERS
MODIFY
fastavro/__init__.py
fastavro/__init__.py
@@ -70,7 +70,6 @@ load = fastavro.read.read_data
writer = fastavro.write.writer
schemaless_writer = fastavro.write.schemaless_writer
dump = fastavro.write.dump
-acquaint_schema = fastavro.schema.acquaint_schema
acquaint_schema = _acquaint_schema
fastavro.schema.acquaint_schema = _acquaint_schema
is_avro = fastavro.read.is_avro

MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -21,9 +21,10 @@ from ._six import (
btou, iteritems, is_str, str2ints, fstint
)
from ._schema import (
- extract_record_type, acquaint_schema, populate_schema_defs,
+ extract_record_type, extract_named_schemas_into_repo, populate_schema_defs,
extract_logical_type
)
+from ._schema_common import SCHEMA_DEFS
from ._read_common import (
SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA
)
@@ -608,6 +609,16 @@ except ImportError:
pass
+def acquaint_schema(schema):
+ """Extract schema into READERS"""
+ extract_named_schemas_into_repo(
+ schema,
+ READERS,
+ lambda schema: lambda fo, _, r_schema: read_data(
+ fo, schema, SCHEMA_DEFS.get(r_schema)),
+ )
+
+
cpdef MemoryReader snappy_read_block(fo):
length = read_long(fo, None)
data = fo.read(length - 4)
@@ -675,7 +686,7 @@ class iter_avro:
self.codec = self.metadata.get('avro.codec', 'null')
self.reader_schema = reader_schema
- acquaint_schema(self.writer_schema, READERS)
+ acquaint_schema(self.writer_schema)
if reader_schema:
populate_schema_defs(reader_schema)
self._records = _iter_avro(fo_reader,
@@ -703,7 +714,7 @@ cpdef schemaless_reader(fo, schema):
schema: dict
Reader schema
"""
- acquaint_schema(schema, READERS)
+ acquaint_schema(schema)
return read_data(fo, schema)

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -16,9 +16,10 @@ from .six import (
MemoryIO, xrange, btou, iteritems, is_str, str2ints, fstint
)
from .schema import (
- extract_record_type, acquaint_schema, populate_schema_defs,
+ extract_record_type, extract_named_schemas_into_repo, populate_schema_defs,
extract_logical_type
)
+from ._schema_common import SCHEMA_DEFS
from ._read_common import (
SchemaResolutionError, MAGIC, SYNC_SIZE, HEADER_SCHEMA,
)
@@ -512,6 +513,16 @@ except ImportError:
pass
+def acquaint_schema(schema):
+ """Extract schema into READERS"""
+ extract_named_schemas_into_repo(
+ schema,
+ READERS,
+ lambda schema: lambda fo, _, r_schema: read_data(
+ fo, schema, SCHEMA_DEFS.get(r_schema)),
+ )
+
+
def _iter_avro(fo, header, codec, writer_schema, reader_schema):
"""Return iterator over avro records."""
sync_marker = header['sync']
@@ -569,7 +580,7 @@ class iter_avro:
self.codec = self.metadata.get('avro.codec', 'null')
self.reader_schema = reader_schema
- acquaint_schema(self.writer_schema, READERS)
+ acquaint_schema(self.writer_schema)
if reader_schema:
populate_schema_defs(reader_schema)
self._records = _iter_avro(fo,
@@ -597,7 +608,7 @@ def schemaless_reader(fo, schema):
schema: dict
Reader schema
"""
- acquaint_schema(schema, READERS)
+ acquaint_schema(schema)
return read_data(fo, schema)

MODIFY
fastavro/_schema.pyx
fastavro/_schema.pyx
@@ -121,48 +121,29 @@ def load_schema(schema_path):
return _load_schema(schema, schema_dir)
-def _read():
- # FIXME: This is due to circular depedency, find a better way
- try:
- from . import _read as read
- except ImportError:
- from . import read
-
- return read
-
-
def _load_schema(schema, schema_dir):
try:
- _read().acquaint_schema(schema)
+ from fastavro import acquaint_schema
+ acquaint_schema(schema)
except UnknownType as e:
try:
avsc = path.join(schema_dir, '%s.avsc' % e.name)
- load_schema(avsc)
+ sub_schema = load_schema(avsc)
except IOError:
raise e
- _load_schema(schema, schema_dir)
+
+ if isinstance(schema, dict):
+ return _load_schema([sub_schema, schema], schema_dir)
+ else:
+ # schema is already a list
+ schema.insert(sub_schema, 0)
+ return _load_schema(schema, schema_dir)
return schema
-def populate_schema_defs(schema, repo=None):
- repo = SCHEMA_DEFS if repo is None else repo
+def populate_schema_defs(schema):
extract_named_schemas_into_repo(
schema,
- repo,
+ SCHEMA_DEFS,
lambda schema: schema,
)
-
-
-def acquaint_schema(schema,
- repo=None,
- reader_schema_defs=None):
- """Extract schema in repo (default READERS)"""
- repo = _read().READERS if repo is None else repo
- reader_schema_defs = \
- SCHEMA_DEFS if reader_schema_defs is None else reader_schema_defs
- extract_named_schemas_into_repo(
- schema,
- repo,
- lambda schema: lambda fo, _, r_schema: _read().read_data(
- fo, schema, reader_schema_defs.get(r_schema)),
- )

MODIFY
fastavro/_schema_py.py
fastavro/_schema_py.py
@@ -120,48 +120,29 @@ def load_schema(schema_path):
return _load_schema(schema, schema_dir)
-def _read():
- # FIXME: This is due to circular depedency, find a better way
- try:
- from . import _read as read
- except ImportError:
- from . import read
-
- return read
-
-
def _load_schema(schema, schema_dir):
try:
- _read().acquaint_schema(schema)
+ from fastavro import acquaint_schema
+ acquaint_schema(schema)
except UnknownType as e:
try:
avsc = path.join(schema_dir, '%s.avsc' % e.name)
- load_schema(avsc)
+ sub_schema = load_schema(avsc)
except IOError:
raise e
- _load_schema(schema, schema_dir)
+
+ if isinstance(schema, dict):
+ return _load_schema([sub_schema, schema], schema_dir)
+ else:
+ # schema is already a list
+ schema.insert(sub_schema, 0)
+ return _load_schema(schema, schema_dir)
return schema
-def populate_schema_defs(schema, repo=None):
- repo = SCHEMA_DEFS if repo is None else repo
+def populate_schema_defs(schema):
extract_named_schemas_into_repo(
schema,
- repo,
+ SCHEMA_DEFS,
lambda schema: schema,
)
-
-
-def acquaint_schema(schema,
- repo=None,
- reader_schema_defs=None):
- """Extract schema in repo (default READERS)"""
- repo = _read().READERS if repo is None else repo
- reader_schema_defs = \
- SCHEMA_DEFS if reader_schema_defs is None else reader_schema_defs
- extract_named_schemas_into_repo(
- schema,
- repo,
- lambda schema: lambda fo, _, r_schema: _read().read_data(
- fo, schema, reader_schema_defs.get(r_schema)),
- )

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
@@ -29,7 +29,7 @@ from ._schema import (
extract_named_schemas_into_repo, extract_record_type,
extract_logical_type
)
-from ._write_common import SCHEMA_DEFS
+from ._schema_common import SCHEMA_DEFS
NoneType = type(None)
@@ -696,12 +696,11 @@ cpdef snappy_write_block(object fo, bytes block_bytes):
fo.write(tmp)
-def acquaint_schema(schema, repo=None):
- """Extract schema into repo (default WRITERS)"""
- repo = WRITERS if repo is None else repo
+def acquaint_schema(schema):
+ """Extract schema into WRITERS"""
extract_named_schemas_into_repo(
schema,
- repo,
+ WRITERS,
lambda schema: lambda bytearray fo, datum, _: write_data(fo, datum, schema),
)
extract_named_schemas_into_repo(

DELETE
fastavro/_write_common.py
-_base_types = [
- 'boolean',
- 'bytes',
- 'double',
- 'float',
- 'int',
- 'long',
- 'null',
- 'string',
-]
-
-SCHEMA_DEFS = {typ: typ for typ in _base_types}

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -31,7 +31,7 @@ from .schema import (
extract_named_schemas_into_repo, extract_record_type,
extract_logical_type
)
-from ._write_common import SCHEMA_DEFS
+from ._schema_common import SCHEMA_DEFS
NoneType = type(None)
@@ -539,12 +539,11 @@ except ImportError:
pass
-def acquaint_schema(schema, repo=None):
- """Extract schema into repo (default WRITERS)"""
- repo = WRITERS if repo is None else repo
+def acquaint_schema(schema):
+ """Extract schema into WRITERS"""
extract_named_schemas_into_repo(
schema,
- repo,
+ WRITERS,
lambda schema: lambda fo, datum, _: write_data(fo, datum, schema),
)
extract_named_schemas_into_repo(

MODIFY
fastavro/schema.py
fastavro/schema.py
@@ -7,12 +7,11 @@ from ._schema_common import UnknownType
load_schema = _schema.load_schema
extract_record_type = _schema.extract_record_type
-acquaint_schema = _schema.acquaint_schema
populate_schema_defs = _schema.populate_schema_defs
extract_logical_type = _schema.extract_logical_type
extract_named_schemas_into_repo = _schema.extract_named_schemas_into_repo
__all__ = [
- 'UnknownType', 'load_schema', 'extract_record_type', 'acquaint_schema',
+ 'UnknownType', 'load_schema', 'extract_record_type',
'extract_logical_type', 'extract_named_schemas_into_repo',
]

MODIFY
fastavro/write.py
fastavro/write.py
@@ -3,8 +3,7 @@ try:
except ImportError as e:
from . import _write_py as _write
-from ._write_common import SCHEMA_DEFS
-
+SCHEMA_DEFS = _write.SCHEMA_DEFS
acquaint_schema = _write.acquaint_schema
dump = _write.dump
writer = _write.writer

ADD
tests/conftest.py
+import pytest
+
+from fastavro.read import READERS
+from fastavro.write import WRITERS
+from fastavro._schema_common import SCHEMA_DEFS
+
+
+@pytest.fixture(scope='function')
+def clean_readers_writers_and_schemas():
+ reader_keys = {key for key in READERS.keys()}
+ writer_keys = {key for key in WRITERS.keys()}
+ schema_keys = {key for key in SCHEMA_DEFS.keys()}
+
+ yield
+
+ repo_keys = (
+ (READERS, reader_keys),
+ (WRITERS, writer_keys),
+ (SCHEMA_DEFS, schema_keys),
+ )
+ for repo, keys in repo_keys:
+ diff = set(repo) - keys
+ for key in diff:
+ del repo[key]

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
import fastavro
from fastavro.read import _read as _reader
from fastavro.write import _write as _writer, Writer
+from fastavro._schema_common import SCHEMA_DEFS
from fastavro.six import MemoryIO
@@ -12,6 +13,7 @@ from collections import OrderedDict
from os.path import join, abspath, dirname, basename
from glob import iglob
+pytestmark = pytest.mark.usefixtures("clean_readers_writers_and_schemas")
data_dir = join(abspath(dirname(__file__)), 'avro-files')
@@ -218,9 +220,32 @@ def test_acquaint_schema_accepts_nested_records_from_arrays():
def test_compose_schemas():
schema_path = join(data_dir, 'Parent.avsc')
- schema = fastavro.schema.load_schema(schema_path)
- assert isinstance(schema, dict)
+ fastavro.schema.load_schema(schema_path)
+ assert 'Parent' in fastavro.read.READERS
assert 'Child' in fastavro.read.READERS
+ assert 'Parent' in fastavro.write.WRITERS
+ assert 'Child' in fastavro.write.WRITERS
+
+
+def test_reading_after_writing_with_load_schema():
+ schema_path = join(data_dir, 'Parent.avsc')
+ schema = fastavro.schema.load_schema(schema_path)
+
+ records = [{'child': {}}]
+
+ new_file = MemoryIO()
+ fastavro.writer(new_file, schema, records)
+ new_file.seek(0)
+
+ # Clean the Child and Parent entries so we are forced to get them from the
+ # schema
+ for repo in (SCHEMA_DEFS, fastavro.write.WRITERS, fastavro.read.READERS):
+ del repo['Child']
+ del repo['Parent']
+
+ reader = fastavro.reader(new_file)
+ new_records = list(reader)
+ assert new_records == records
def test_missing_schema():

ensure acquaint_schema acquints to both READERS and WRITERS