change load_schema so that it does not create union schemas unless the initially starting point is a union schema
MODIFY
fastavro/_schema.pyx
fastavro/_schema.pyx
@@ -277,42 +277,104 @@ cdef parse_field(field, namespace, expand, names, named_schemas):
return parsed_field
-def load_schema(schema_path, _named_schemas=None):
+def load_schema(schema_path, _named_schemas=None, _write_hint=True):
if _named_schemas is None:
_named_schemas = {}
with open(schema_path) as fd:
schema = json.load(fd)
schema_dir, schema_file = path.split(schema_path)
- return _load_schema(schema, schema_dir, _named_schemas)
+ return _load_schema(schema, schema_dir, _named_schemas, _write_hint)
-cdef _load_schema(schema, schema_dir, named_schemas):
+cdef _load_schema(schema, schema_dir, named_schemas, write_hint):
try:
schema_copy = deepcopy(named_schemas)
- return parse_schema(schema, _named_schemas=named_schemas)
+ return parse_schema(
+ schema, _named_schemas=named_schemas, _write_hint=write_hint
+ )
except UnknownType as e:
try:
avsc = path.join(schema_dir, '%s.avsc' % e.name)
- sub_schema = load_schema(avsc, _named_schemas=schema_copy)
+ sub_schema = load_schema(
+ avsc, _named_schemas=schema_copy, _write_hint=False
+ )
except IOError:
raise e
- if isinstance(schema, dict):
- if isinstance(sub_schema, list):
- return _load_schema(
- sub_schema + [schema], schema_dir, schema_copy
+ _inject_schema(schema, sub_schema)
+ return _load_schema(schema, schema_dir, schema_copy, write_hint)
+
+
+cdef _inject_schema(outer_schema, inner_schema, is_injected=False):
+ # Once injected, we can stop checking to see if we need to inject since it
+ # should only be done once at most
+ if is_injected is True:
+ return outer_schema, is_injected
+
+ # union schemas
+ if isinstance(outer_schema, list):
+ union = []
+ for each_schema in outer_schema:
+ if is_injected:
+ union.append(each_schema)
+ else:
+ return_schema, injected = _inject_schema(
+ each_schema, inner_schema, is_injected
)
+ union.append(return_schema)
+ if injected is True:
+ is_injected = injected
+ return union, is_injected
+
+ # string schemas; this could be either a named schema or a primitive type
+ elif not isinstance(outer_schema, dict):
+ if outer_schema in PRIMITIVES:
+ return outer_schema, is_injected
+ elif outer_schema == inner_schema["name"]:
+ return inner_schema, True
else:
- return _load_schema(
- [sub_schema, schema], schema_dir, schema_copy
+ raise Exception(
+ "Internal error; "
+ + "You should raise an issue in the fastavro github repository"
)
else:
- # schema is already a list
- if isinstance(sub_schema, list):
- return _load_schema(
- sub_schema + schema, schema_dir, schema_copy
+ # Remaining valid schemas must be dict types
+ schema_type = outer_schema["type"]
+
+ if schema_type == "array":
+ return_schema, injected = _inject_schema(
+ outer_schema["items"], inner_schema, is_injected
)
+ outer_schema["items"] = return_schema
+ return outer_schema, injected
+
+ elif schema_type == "map":
+ return_schema, injected = _inject_schema(
+ outer_schema["values"], inner_schema, is_injected
+ )
+ outer_schema["values"] = return_schema
+ return outer_schema, injected
+
+ elif schema_type == "record" or schema_type == "error":
+ # records
+ fields = []
+ for field in outer_schema.get('fields', []):
+ if is_injected:
+ fields.append(field)
else:
- schema.insert(0, sub_schema)
- return _load_schema(schema, schema_dir, schema_copy)
+ return_schema, injected = _inject_schema(
+ field["type"], inner_schema, is_injected
+ )
+ field["type"] = return_schema
+ fields.append(field)
+
+ if injected is True:
+ is_injected = injected
+ if fields:
+ outer_schema["fields"] = fields
+
+ return outer_schema, is_injected
+
+ elif schema_type in PRIMITIVES:
+ return outer_schema, is_injected

MODIFY
fastavro/_schema_py.py
fastavro/_schema_py.py
@@ -410,7 +410,7 @@ def parse_field(field, namespace, expand, names, named_schemas):
return parsed_field
-def load_schema(schema_path, *, _named_schemas=None):
+def load_schema(schema_path, *, _named_schemas=None, _write_hint=True):
"""Returns a schema loaded from the file at `schema_path`.
Will recursively load referenced schemas assuming they can be found in
@@ -424,6 +424,9 @@ def load_schema(schema_path, *, _named_schemas=None):
_named_schemas: dict
Internal API argument. Dictionary of named schemas to their schema
definition
+ _write_hint: bool
+ Internal API argument specifying whether or not the __fastavro_parsed
+ marker should be added to the schema
Consider the following example...
@@ -465,35 +468,97 @@ def load_schema(schema_path, *, _named_schemas=None):
with open(schema_path) as fd:
schema = json.load(fd)
schema_dir, schema_file = path.split(schema_path)
- return _load_schema(schema, schema_dir, _named_schemas)
+ return _load_schema(schema, schema_dir, _named_schemas, _write_hint)
-def _load_schema(schema, schema_dir, named_schemas):
+def _load_schema(schema, schema_dir, named_schemas, write_hint):
try:
schema_copy = deepcopy(named_schemas)
- return parse_schema(schema, _named_schemas=named_schemas)
+ return parse_schema(
+ schema, _named_schemas=named_schemas, _write_hint=write_hint
+ )
except UnknownType as e:
try:
avsc = path.join(schema_dir, '%s.avsc' % e.name)
- sub_schema = load_schema(avsc, _named_schemas=schema_copy)
+ sub_schema = load_schema(
+ avsc, _named_schemas=schema_copy, _write_hint=False
+ )
except IOError:
raise e
- if isinstance(schema, dict):
- if isinstance(sub_schema, list):
- return _load_schema(
- sub_schema + [schema], schema_dir, schema_copy
+ _inject_schema(schema, sub_schema)
+ return _load_schema(schema, schema_dir, schema_copy, write_hint)
+
+
+def _inject_schema(outer_schema, inner_schema, is_injected=False):
+ # Once injected, we can stop checking to see if we need to inject since it
+ # should only be done once at most
+ if is_injected is True:
+ return outer_schema, is_injected
+
+ # union schemas
+ if isinstance(outer_schema, list):
+ union = []
+ for each_schema in outer_schema:
+ if is_injected:
+ union.append(each_schema)
+ else:
+ return_schema, injected = _inject_schema(
+ each_schema, inner_schema, is_injected
)
+ union.append(return_schema)
+ if injected is True:
+ is_injected = injected
+ return union, is_injected
+
+ # string schemas; this could be either a named schema or a primitive type
+ elif not isinstance(outer_schema, dict):
+ if outer_schema in PRIMITIVES:
+ return outer_schema, is_injected
+ elif outer_schema == inner_schema["name"]:
+ return inner_schema, True
else:
- return _load_schema(
- [sub_schema, schema], schema_dir, schema_copy
+ raise Exception(
+ "Internal error; "
+ + "You should raise an issue in the fastavro github repository"
)
else:
- # schema is already a list
- if isinstance(sub_schema, list):
- return _load_schema(
- sub_schema + schema, schema_dir, schema_copy
+ # Remaining valid schemas must be dict types
+ schema_type = outer_schema["type"]
+
+ if schema_type == "array":
+ return_schema, injected = _inject_schema(
+ outer_schema["items"], inner_schema, is_injected
)
+ outer_schema["items"] = return_schema
+ return outer_schema, injected
+
+ elif schema_type == "map":
+ return_schema, injected = _inject_schema(
+ outer_schema["values"], inner_schema, is_injected
+ )
+ outer_schema["values"] = return_schema
+ return outer_schema, injected
+
+ elif schema_type == "record" or schema_type == "error":
+ # records
+ fields = []
+ for field in outer_schema.get('fields', []):
+ if is_injected:
+ fields.append(field)
else:
- schema.insert(0, sub_schema)
- return _load_schema(schema, schema_dir, schema_copy)
+ return_schema, injected = _inject_schema(
+ field["type"], inner_schema, is_injected
+ )
+ field["type"] = return_schema
+ fields.append(field)
+
+ if injected is True:
+ is_injected = injected
+ if fields:
+ outer_schema["fields"] = fields
+
+ return outer_schema, is_injected
+
+ elif schema_type in PRIMITIVES:
+ return outer_schema, is_injected

ADD
tests/load_schema_test_3/A.avsc
+{
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {"name": "b", "type": ["null", "B"], "default": null},
+ {"name": "c", "type": ["null", "C"], "default": null}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_3/B.avsc
+{
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {"name": "c", "type": ["null", "C"], "default": null}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_3/C.avsc
+{
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_4/A.avsc
+{
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {"name": "b", "type": ["null", "B"], "default": null},
+ {"name": "c", "type": ["null", "C"], "default": null}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_4/B.avsc
+{
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_4/C.avsc
+{
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {"name": "b", "type": ["null", "B"], "default": null}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_5/A.avsc
+{
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {"name": "map_field", "type": {"type": "map", "values": "B"}},
+ {"name": "b", "type": "B"}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_5/B.avsc
+{
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_6/A.avsc
+{
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {"name": "array_field", "type": {"type": "array", "items": "B"}},
+ {"name": "b", "type": "B"}
+ ]
+}
\ No newline at end of file

ADD
tests/load_schema_test_6/B.avsc
+{
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+}
\ No newline at end of file

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -251,24 +251,6 @@ def test_missing_schema():
fastavro.schema.load_schema(schema_path)
-def test_load_schema_does_not_make_unions_of_unions():
- """https://github.com/fastavro/fastavro/issues/443"""
- load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test')
- schema_path = join(load_schema_dir, 'A.avsc')
- loaded_schema = fastavro.schema.load_schema(schema_path)
- for schema in loaded_schema:
- assert not isinstance(schema, list)
-
-
-def test_load_schema_does_not_make_unions_of_unions_2():
- """https://github.com/fastavro/fastavro/issues/443"""
- load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_2')
- schema_path = join(load_schema_dir, 'A.avsc')
- loaded_schema = fastavro.schema.load_schema(schema_path)
- for schema in loaded_schema:
- assert not isinstance(schema, list)
-
-
def test_default_values():
schema = {
"type": "record",

MODIFY
tests/test_schema.py
tests/test_schema.py
+from os.path import join, abspath, dirname
import pytest
import fastavro
from fastavro.schema import (
@@ -524,3 +525,219 @@ def test_union_schemas_must_have_names_in_order():
# This should not work because Location is defined after it is used
with pytest.raises(UnknownType):
parse_schema(schema2)
+
+
+def test_load_schema_does_not_make_unions_of_unions():
+ """https://github.com/fastavro/fastavro/issues/443"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path)
+ assert isinstance(loaded_schema, dict)
+
+
+def test_load_schema_does_not_make_unions_of_unions_2():
+ """https://github.com/fastavro/fastavro/issues/443"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_2')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path)
+ assert isinstance(loaded_schema, list)
+ for schema in loaded_schema:
+ assert not isinstance(schema, list)
+
+
+def test_load_schema_output_is_correct():
+ """https://github.com/fastavro/fastavro/issues/476"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path, _write_hint=False)
+
+ expected_schema = {
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {
+ "name": "b",
+ "type": [
+ "null",
+ {
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {
+ "name": "c",
+ "type": [
+ "null",
+ {
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+ },
+ ],
+ "default": None,
+ }
+ ]
+ },
+ ],
+ "default": None,
+ },
+ {
+ "name": "d",
+ "type": [
+ "null",
+ {
+ "name": "D",
+ "type": "record",
+ "fields": [
+ {"name": "bar", "type": "string"}
+ ]
+ },
+ ],
+ "default": None,
+ }
+ ]
+ }
+ assert loaded_schema == expected_schema
+
+
+def test_load_schema_output_is_correct_3():
+ """https://github.com/fastavro/fastavro/issues/476"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_3')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path, _write_hint=False)
+
+ expected_schema = {
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {
+ "name": "b",
+ "type": [
+ "null",
+ {
+ "name": "B",
+ "type": "record",
+ "fields": [
+ {
+ "name": "c",
+ "type": [
+ "null",
+ {
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {"name": "foo", "type": "string"}
+ ]
+ },
+ ],
+ "default": None,
+ }
+ ]
+ },
+ ],
+ "default": None,
+ },
+ {"name": "c", "type": ["null", "C"], "default": None}
+ ]
+ }
+ assert loaded_schema == expected_schema
+
+
+def test_load_schema_output_is_correct_4():
+ """https://github.com/fastavro/fastavro/issues/476"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_4')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path, _write_hint=False)
+
+ expected_schema = {
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {
+ "name": "b",
+ "type": [
+ "null",
+ {
+ "name": "B",
+ "type": "record",
+ "fields": [{"name": "foo", "type": "string"}],
+ },
+ ],
+ "default": None,
+ },
+ {
+ "name": "c",
+ "type": [
+ "null",
+ {
+ "name": "C",
+ "type": "record",
+ "fields": [
+ {
+ "name": "b",
+ "type": ["null", "B"],
+ "default": None,
+ },
+ ]
+ },
+ ],
+ "default": None,
+ }
+ ]
+ }
+ assert loaded_schema == expected_schema
+
+
+def test_load_schema_output_is_correct_5():
+ """https://github.com/fastavro/fastavro/issues/476"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_5')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path, _write_hint=False)
+
+ expected_schema = {
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {
+ "name": "map_field",
+ "type": {
+ "type": "map",
+ "values": {
+ "name": "B",
+ "type": "record",
+ "fields": [{"name": "foo", "type": "string"}],
+ },
+ },
+ },
+ {"name": "b", "type": "B"},
+ ]
+ }
+ assert loaded_schema == expected_schema
+
+
+def test_load_schema_output_is_correct_6():
+ """https://github.com/fastavro/fastavro/issues/476"""
+ load_schema_dir = join(abspath(dirname(__file__)), 'load_schema_test_6')
+ schema_path = join(load_schema_dir, 'A.avsc')
+ loaded_schema = fastavro.schema.load_schema(schema_path, _write_hint=False)
+
+ expected_schema = {
+ "name": "A",
+ "type": "record",
+ "fields": [
+ {
+ "name": "array_field",
+ "type": {
+ "type": "array",
+ "items": {
+ "name": "B",
+ "type": "record",
+ "fields": [{"name": "foo", "type": "string"}],
+ },
+ },
+ },
+ {"name": "b", "type": "B"},
+ ]
+ }
+ assert loaded_schema == expected_schema

change load_schema so that it does not create union schemas unless the initially starting point is a union schema