make named_schemas a public argument
MODIFY
fastavro/_read.pyx
fastavro/_read.pyx
@@ -1098,18 +1098,13 @@ class file_reader:
self._named_schemas = {}
if reader_schema:
self.reader_schema = parse_schema(
- reader_schema,
- _write_hint=False,
- _named_schemas=self._named_schemas,
+ reader_schema, self._named_schemas, _write_hint=False
)
else:
self.reader_schema = None
self.writer_schema = parse_schema(
- self._schema,
- _write_hint=False,
- _force=True,
- _named_schemas=self._named_schemas,
+ self._schema, self._named_schemas, _write_hint=False, _force=True
)
self._elems = None
@@ -1167,7 +1162,7 @@ cpdef schemaless_reader(fo, writer_schema, reader_schema=None,
reader_schema = None
named_schemas = {}
- writer_schema = parse_schema(writer_schema, _named_schemas=named_schemas)
+ writer_schema = parse_schema(writer_schema, named_schemas)
if reader_schema:
reader_schema = parse_schema(reader_schema)

MODIFY
fastavro/_read_py.py
fastavro/_read_py.py
@@ -947,9 +947,7 @@ class file_reader:
self._named_schemas = {}
if reader_schema:
self.reader_schema = parse_schema(
- reader_schema,
- _write_hint=False,
- _named_schemas=self._named_schemas,
+ reader_schema, self._named_schemas, _write_hint=False
)
else:
self.reader_schema = None
@@ -977,10 +975,7 @@ class file_reader:
# Always parse the writer schema since it might have named types that
# need to be stored in self._named_types
self.writer_schema = parse_schema(
- self._schema,
- _write_hint=False,
- _force=True,
- _named_schemas=self._named_schemas,
+ self._schema, self._named_schemas, _write_hint=False, _force=True
)
@property
@@ -1180,7 +1175,7 @@ def schemaless_reader(fo, writer_schema, reader_schema=None, return_record_name=
reader_schema = None
named_schemas = {}
- writer_schema = parse_schema(writer_schema, _named_schemas=named_schemas)
+ writer_schema = parse_schema(writer_schema, named_schemas)
if reader_schema:
reader_schema = parse_schema(reader_schema)

MODIFY
fastavro/_schema.pyi
fastavro/_schema.pyi
@@ -7,10 +7,11 @@ FINGERPRINT_ALGORITHMS: str
def load_schema(schema_path: str) -> Dict: ...
def parse_schema(
schema: SchemaType,
+ named_schemas: Dict = ...,
+ *,
expand: bool = ...,
_write_hint: bool = ...,
_force: bool = ...,
- _named_schemas: Dict = ...,
) -> SchemaType: ...
def fullname(schema: Dict) -> str: ...
def expand_schema(schema: Dict) -> Dict: ...

MODIFY
fastavro/_schema.pyx
fastavro/_schema.pyx
@@ -69,29 +69,31 @@ cpdef expand_schema(schema):
def parse_schema(
- schema, expand=False, _write_hint=True, _force=False, _named_schemas=None
+ schema, named_schemas=None, *, expand=False, _write_hint=True, _force=False
):
- if _named_schemas is None:
- _named_schemas = {}
+ if named_schemas is None:
+ named_schemas = {}
if _force or expand:
return _parse_schema(
- schema, "", expand, _write_hint, set(), _named_schemas
+ schema, "", expand, _write_hint, set(), named_schemas
)
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
for key, value in schema["__named_schemas"].items():
- _named_schemas[key] = value
+ named_schemas[key] = value
return schema
elif isinstance(schema, list):
# If we are given a list we should make sure that the immediate sub
# schemas have the hint in them
return [
- parse_schema(s, expand, _write_hint, _force, _named_schemas)
+ parse_schema(
+ s, named_schemas, expand=expand, _write_hint=_write_hint, _force=_force
+ )
for s in schema
]
else:
return _parse_schema(
- schema, "", expand, _write_hint, set(), _named_schemas
+ schema, "", expand, _write_hint, set(), named_schemas
)
@@ -280,10 +282,10 @@ cdef parse_field(field, namespace, expand, names, named_schemas):
def load_schema(
- schema_path, *, _named_schemas=None, _write_hint=True, _injected_schemas=None
+ schema_path, *, named_schemas=None, _write_hint=True, _injected_schemas=None
):
- if _named_schemas is None:
- _named_schemas = {}
+ if named_schemas is None:
+ named_schemas = {}
if _injected_schemas is None:
_injected_schemas = set()
@@ -292,22 +294,20 @@ def load_schema(
schema = json.load(fd)
schema_dir, schema_file = path.split(schema_path)
return _load_schema(
- schema, schema_dir, _named_schemas, _write_hint, _injected_schemas
+ schema, schema_dir, named_schemas, _write_hint, _injected_schemas
)
cdef _load_schema(schema, schema_dir, named_schemas, write_hint, injected_schemas):
try:
schema_copy = deepcopy(named_schemas)
- return parse_schema(
- schema, _named_schemas=named_schemas, _write_hint=write_hint
- )
+ return parse_schema(schema, named_schemas=named_schemas, _write_hint=write_hint)
except UnknownType as e:
try:
avsc = path.join(schema_dir, f"{e.name}.avsc")
sub_schema = load_schema(
avsc,
- _named_schemas=schema_copy,
+ named_schemas=schema_copy,
_write_hint=False,
_injected_schemas=injected_schemas,
)
@@ -414,12 +414,12 @@ cdef _inject_schema(outer_schema, inner_schema, namespace="", is_injected=False)
def load_schema_ordered(ordered_schemas, *, _write_hint=True):
loaded_schemas = []
- _named_schemas = {}
+ named_schemas = {}
for idx, schema_path in enumerate(ordered_schemas):
# _write_hint is always False except maybe the outer most schema
_last = _write_hint if idx + 1 == len(ordered_schemas) else False
schema = load_schema(
- schema_path, _named_schemas=_named_schemas, _write_hint=_last
+ schema_path, named_schemas=named_schemas, _write_hint=_last
)
loaded_schemas.append(schema)

MODIFY
fastavro/_schema_py.py
fastavro/_schema_py.py
@@ -160,10 +160,11 @@ def expand_schema(schema):
def parse_schema(
schema,
+ named_schemas=None,
+ *,
expand=False,
_write_hint=True,
_force=False,
- _named_schemas=None,
):
"""Returns a parsed avro schema
@@ -175,10 +176,9 @@ def parse_schema(
----------
schema: dict
Input schema
+ named_schemas: dict
+ Dictionary of named schemas to their schema definition
expand: bool
- NOTE: This option should be considered a keyword only argument and may
- get enforced as such when Python 2 support is dropped.
-
If true, named schemas will be fully expanded to their true schemas
rather than being represented as just the name. This format should be
considered an output only and not passed in to other reader/writer
@@ -190,9 +190,6 @@ def parse_schema(
_force: bool
Internal API argument. If True, the schema will always be parsed even
if it has been parsed and has the __fastavro_parsed marker
- _named_schemas: dict
- Internal API argument. Dictionary of named schemas to their schema
- definition
Example::
@@ -203,24 +200,43 @@ def parse_schema(
parsed_schema = parse_schema(original_schema)
with open('weather.avro', 'wb') as out:
writer(out, parsed_schema, records)
+
+
+ Sometimes you might have two schemas where one schema references another.
+ For the sake of example, let's assume you have a `Parent` schema that
+ references a `Child` schema`. If you were to try to parse the parent schema
+ on its own, you would get an exception because the child schema isn't
+ defined. To accomodate this, we can use the `named_schemas` argument to pass
+ a shared dictionary when parsing both of the schemas. The dictionary will
+ get populated with the necessary schema references to make parsing possible.
+ For example::
+
+ from fastavro import parse_schema
+
+ named_schemas = {}
+ parsed_child = parse_schema(child_schema, named_schemas)
+ parsed_parent = parse_schema(parent_schema, named_schemas)
"""
- if _named_schemas is None:
- _named_schemas = {}
+ if named_schemas is None:
+ named_schemas = {}
if _force or expand:
- return _parse_schema(schema, "", expand, _write_hint, set(), _named_schemas)
+ return _parse_schema(schema, "", expand, _write_hint, set(), named_schemas)
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
for key, value in schema["__named_schemas"].items():
- _named_schemas[key] = value
+ named_schemas[key] = value
return schema
elif isinstance(schema, list):
# If we are given a list we should make sure that the immediate sub
# schemas have the hint in them
return [
- parse_schema(s, expand, _write_hint, _force, _named_schemas) for s in schema
+ parse_schema(
+ s, named_schemas, expand=expand, _write_hint=_write_hint, _force=_force
+ )
+ for s in schema
]
else:
- return _parse_schema(schema, "", expand, _write_hint, set(), _named_schemas)
+ return _parse_schema(schema, "", expand, _write_hint, set(), named_schemas)
def _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas):
@@ -404,7 +420,7 @@ def parse_field(field, namespace, expand, names, named_schemas):
def load_schema(
- schema_path, *, _named_schemas=None, _write_hint=True, _injected_schemas=None
+ schema_path, *, named_schemas=None, _write_hint=True, _injected_schemas=None
):
"""Returns a schema loaded from the file at `schema_path`.
@@ -416,9 +432,8 @@ def load_schema(
----------
schema: str
Path to schema file to load
- _named_schemas: dict
- Internal API argument. Dictionary of named schemas to their schema
- definition
+ named_schemas: dict
+ 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
@@ -460,8 +475,8 @@ def load_schema(
parsed_schema = load_schema("Parent.avsc")
"""
- if _named_schemas is None:
- _named_schemas = {}
+ if named_schemas is None:
+ named_schemas = {}
if _injected_schemas is None:
_injected_schemas = set()
@@ -470,22 +485,20 @@ def load_schema(
schema = json.load(fd)
schema_dir, schema_file = path.split(schema_path)
return _load_schema(
- schema, schema_dir, _named_schemas, _write_hint, _injected_schemas
+ schema, schema_dir, named_schemas, _write_hint, _injected_schemas
)
def _load_schema(schema, schema_dir, named_schemas, write_hint, injected_schemas):
try:
schema_copy = deepcopy(named_schemas)
- return parse_schema(
- schema, _named_schemas=named_schemas, _write_hint=write_hint
- )
+ return parse_schema(schema, named_schemas=named_schemas, _write_hint=write_hint)
except UnknownType as e:
try:
avsc = path.join(schema_dir, f"{e.name}.avsc")
sub_schema = load_schema(
avsc,
- _named_schemas=schema_copy,
+ named_schemas=schema_copy,
_write_hint=False,
_injected_schemas=injected_schemas,
)
@@ -644,12 +657,12 @@ def load_schema_ordered(ordered_schemas, *, _write_hint=True):
)
"""
loaded_schemas = []
- _named_schemas = {}
+ named_schemas = {}
for idx, schema_path in enumerate(ordered_schemas):
# _write_hint is always False except maybe the outer most schema
_last = _write_hint if idx + 1 == len(ordered_schemas) else False
schema = load_schema(
- schema_path, _named_schemas=_named_schemas, _write_hint=_last
+ schema_path, named_schemas=named_schemas, _write_hint=_last
)
loaded_schemas.append(schema)

MODIFY
fastavro/_validation.pyx
fastavro/_validation.pyx
@@ -259,7 +259,7 @@ cpdef _validate(
cpdef validate(object datum, object schema, str field="",
bint raise_errors=True):
named_schemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
return _validate(datum, parsed_schema, named_schemas, field, raise_errors)
@@ -268,7 +268,7 @@ cpdef validate_many(records, schema, bint raise_errors=True):
cdef list errors = []
cdef list results = []
named_schemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
for record in records:
try:
result = _validate(

MODIFY
fastavro/_validation_py.py
fastavro/_validation_py.py
@@ -393,7 +393,7 @@ def validate(datum, schema, field=None, raise_errors=True):
validate(record, schema)
"""
named_schemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
return _validate(datum, parsed_schema, named_schemas, field, raise_errors)
@@ -420,7 +420,7 @@ def validate_many(records, schema, raise_errors=True):
validate_many(records, schema)
"""
named_schemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
errors = []
results = []
for record in records:

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
@@ -598,7 +598,7 @@ cdef class Writer:
self.fo = fo
self._named_schemas = {}
- self.schema = parse_schema(schema, _named_schemas=self._named_schemas)
+ self.schema = parse_schema(schema, self._named_schemas)
self.validate_fn = _validate if validator is True else validator
self.io = MemoryIO()
self.block_count = 0
@@ -730,6 +730,6 @@ def writer(fo,
def schemaless_writer(fo, schema, record):
cdef bytearray tmp = bytearray()
named_schemas = {}
- schema = parse_schema(schema, _named_schemas=named_schemas)
+ schema = parse_schema(schema, named_schemas)
write_data(tmp, record, schema, named_schemas, "")
fo.write(tmp)

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -397,7 +397,7 @@ else:
class GenericWriter:
def __init__(self, schema, metadata=None, validator=None):
self._named_schemas = {}
- self.schema = parse_schema(schema, _named_schemas=self._named_schemas)
+ self.schema = parse_schema(schema, self._named_schemas)
self.validate_fn = _validate if validator is True else validator
self.metadata = metadata or {}
@@ -680,7 +680,7 @@ def schemaless_writer(fo, schema, record):
Note: The ``schemaless_writer`` can only write a single record.
"""
named_schemas = {}
- schema = parse_schema(schema, _named_schemas=named_schemas)
+ schema = parse_schema(schema, named_schemas)
encoder = BinaryEncoder(fo)
write_data(encoder, record, schema, named_schemas, "")

MODIFY
fastavro/utils.py
fastavro/utils.py
@@ -142,7 +142,7 @@ def generate_many(schema: Schema, count: int) -> Iterator[Any]:
writer(out, schema, generate_data(schema, 5))
"""
named_schemas: NamedSchemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
for index in range(count):
yield gen_data(parsed_schema, named_schemas, index)
@@ -163,7 +163,7 @@ def anonymize_schema(schema: Schema) -> Schema:
anonymized_schema = anonymize_schema(original_schema)
"""
named_schemas: NamedSchemas = {}
- parsed_schema = parse_schema(schema, _named_schemas=named_schemas)
+ parsed_schema = parse_schema(schema, named_schemas)
return _anonymize_schema(parsed_schema, named_schemas)

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -2253,9 +2253,7 @@ def test_reading_with_skip_using_cython():
]
named_schemas = {}
- parsed_writer_schema = fastavro.parse_schema(
- writer_schema, _named_schemas=named_schemas
- )
+ parsed_writer_schema = fastavro.parse_schema(writer_schema, named_schemas)
roundtrip_records = roundtrip(
parsed_writer_schema, records, reader_schema=reader_schema
)
@@ -2429,9 +2427,7 @@ def test_reading_with_skip_using_pure_python():
]
named_schemas = {}
- parsed_writer_schema = fastavro.parse_schema(
- writer_schema, _named_schemas=named_schemas
- )
+ parsed_writer_schema = fastavro.parse_schema(writer_schema, named_schemas)
roundtrip_records = roundtrip(
parsed_writer_schema, records, reader_schema=reader_schema
)

MODIFY
tests/test_schema.py
tests/test_schema.py
@@ -541,6 +541,27 @@ def test_union_schemas_must_have_names_in_order():
parse_schema(schema2)
+def test_using_named_schemas_to_handle_references():
+ location = {
+ "name": "Location",
+ "type": "record",
+ "fields": [{"name": "city", "type": "long"}],
+ }
+ weather = {
+ "name": "Weather",
+ "type": "record",
+ "fields": [{"name": "of", "type": "Location"}],
+ }
+
+ named_schemas = {}
+ parse_schema(location, named_schemas)
+ parse_schema(weather, named_schemas)
+
+ # This should not work because didn't supply the named schemas
+ with pytest.raises(UnknownType):
+ parse_schema(weather)
+
+
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")

MODIFY
tests/test_validation.py
tests/test_validation.py
@@ -580,12 +580,12 @@ def test_validate_should_not_parse_schema_if_it_was_parsed_already():
named_schemas = {}
parse_schema(
{"name": "B", "type": "record", "fields": [{"name": "bar", "type": "string"}]},
- _named_schemas=named_schemas,
+ named_schemas,
)
a_schema = parse_schema(
{"name": "A", "type": "record", "fields": [{"name": "b", "type": "B"}]},
- _named_schemas=named_schemas,
+ named_schemas,
)
records = [{"b": {"bar": "bar"}}]

make named_schemas a public argument