add type hint to records
MODIFY
docs/writer.rst
docs/writer.rst
@@ -84,3 +84,40 @@ dictionary. For example::
]
}
]
+
+Using the record hint to specify which branch of a union to take
+----------------------------------------------------------------
+
+In addition to the tuple notation for specifing the name of a record, you can
+also include a special `-type` attribute (note that this attribute is `-type`,
+not `type`) on a record to do the same thing. So the example above which looked
+like this::
+
+ records = [
+ {
+ "address": "123 Drive Street",
+ "family_members": [
+ ("Child", {"name": "Son"}),
+ ("Child", {"name": "Daughter"}),
+ ("Pet", {"name": "Dog"}),
+ ]
+ }
+ ]
+
+Would now look like this::
+
+ records = [
+ {
+ "address": "123 Drive Street",
+ "family_members": [
+ {"-type": "Child", name": "Son"},
+ {"-type": "Child", name": "Daughter"},
+ {"-type": "Pet", name": "Dog"},
+ ]
+ }
+ ]
+
+Unlike the tuple notation which can be used with any avro type in a union, this
+`-type` hint can only be used with records. However, this can be useful if you
+want to make a single record dictionary that can be used both in and out of
+unions.

MODIFY
fastavro/_validation.pyx
fastavro/_validation.pyx
@@ -124,12 +124,14 @@ cdef inline bint validate_record(
) except -1:
if not isinstance(datum, Mapping):
return False
- _, namespace = schema_name(schema, parent_ns)
+ _, fullname = schema_name(schema, parent_ns)
+ if "-type" in datum and datum["-type"] != fullname:
+ return False
for f in schema["fields"]:
if not _validate(datum=datum.get(f["name"], f.get("default")),
schema=f["type"],
named_schemas=named_schemas,
- field=f"{namespace}.{f['name']}",
+ field=f"{fullname}.{f['name']}",
raise_errors=raise_errors):
return False
return True

MODIFY
fastavro/_validation_py.py
fastavro/_validation_py.py
@@ -243,17 +243,21 @@ def validate_record(datum, schema, named_schemas, parent_ns=None, raise_errors=T
raise_errors: bool
If true, raises ValidationError on invalid data
"""
- _, namespace = schema_name(schema, parent_ns)
- return isinstance(datum, Mapping) and all(
+ _, fullname = schema_name(schema, parent_ns)
+ return (
+ isinstance(datum, Mapping)
+ and not ("-type" in datum and datum["-type"] != fullname)
+ and all(
_validate(
datum=datum.get(f["name"], f.get("default")),
schema=f["type"],
named_schemas=named_schemas,
- field=f"{namespace}.{f['name']}",
+ field=f"{fullname}.{f['name']}",
raise_errors=raise_errors,
)
for f in schema["fields"]
)
+ )
def validate_union(datum, schema, named_schemas, parent_ns=None, raise_errors=True):

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -2742,3 +2742,95 @@ def test_tuple_writer_picks_correct_union_path_fixed():
assert expected_roundtrip_value == roundtrip(
parsed_schema, records, return_record_name=True
)
+
+
+@pytest.mark.parametrize(
+ "input_records,expected_roundtrip",
+ [
+ (
+ [
+ {
+ "my_1st_union": {"some_field": 1, "-type": "foo"},
+ "my_2nd_union": {"some_field": 2, "-type": "foo"},
+ }
+ ],
+ [
+ {
+ "my_1st_union": ("foo", {"some_field": 1}),
+ "my_2nd_union": ("foo", {"some_field": 2}),
+ }
+ ],
+ ),
+ (
+ [
+ {
+ "my_1st_union": {"some_field": 3, "-type": "foo"},
+ "my_2nd_union": {"some_field": 4, "-type": "bar"},
+ }
+ ],
+ [
+ {
+ "my_1st_union": ("foo", {"some_field": 3}),
+ "my_2nd_union": ("bar", {"some_field": 4}),
+ }
+ ],
+ ),
+ (
+ [
+ {
+ "my_1st_union": {"some_field": 5, "-type": "bar"},
+ "my_2nd_union": {"some_field": 6, "-type": "foo"},
+ }
+ ],
+ [
+ {
+ "my_1st_union": ("bar", {"some_field": 5}),
+ "my_2nd_union": ("foo", {"some_field": 6}),
+ }
+ ],
+ ),
+ (
+ [
+ {
+ "my_1st_union": {"some_field": 7, "-type": "bar"},
+ "my_2nd_union": {"some_field": 8, "-type": "bar"},
+ }
+ ],
+ [
+ {
+ "my_1st_union": ("bar", {"some_field": 7}),
+ "my_2nd_union": ("bar", {"some_field": 8}),
+ }
+ ],
+ ),
+ ],
+)
+def test_union_path_picked_with_record_type_hint(input_records, expected_roundtrip):
+ """https://github.com/fastavro/fastavro/issues/540"""
+ schema = {
+ "type": "record",
+ "name": "my_record",
+ "fields": [
+ {
+ "name": "my_1st_union",
+ "type": [
+ {
+ "name": "foo",
+ "type": "record",
+ "fields": [{"name": "some_field", "type": "int"}],
+ },
+ {
+ "name": "bar",
+ "type": "record",
+ "fields": [{"name": "some_field", "type": "int"}],
+ },
+ ],
+ },
+ {"name": "my_2nd_union", "type": ["foo", "bar"]},
+ ],
+ }
+
+ rt_records = roundtrip(
+ fastavro.parse_schema(schema), input_records, return_record_name=True
+ )
+ assert expected_roundtrip == rt_records

add type hint to records