add default value checks when parsing the schema
MODIFY
fastavro/_schema.pyx
fastavro/_schema.pyx
@@ -25,6 +25,7 @@ from ._schema_common import (
)
SYMBOL_REGEX = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
+NO_DEFAULT = object()
cpdef inline is_nullable_union(schema):
@@ -97,11 +98,13 @@ def parse_schema(
# Some old schemas might only have __fastavro_parsed and not
# __named_schemas since that came later. For these schemas, we need
# to re-parse the schema to handle named types
- return _parse_schema(schema, "", expand, _write_hint, set(), named_schemas)
+ return _parse_schema(
+ schema, "", expand, _write_hint, set(), named_schemas, NO_DEFAULT
+ )
if _force or expand:
return _parse_schema(
- schema, "", expand, _write_hint, set(), named_schemas
+ schema, "", expand, _write_hint, set(), named_schemas, NO_DEFAULT
)
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
return schema
@@ -116,22 +119,69 @@ def parse_schema(
]
else:
return _parse_schema(
- schema, "", expand, _write_hint, set(), named_schemas
+ schema, "", expand, _write_hint, set(), named_schemas, NO_DEFAULT
)
-cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas):
+cdef _raise_default_value_error(default, schema_type, in_union):
+ if in_union:
+ text = f"first schema in union with type: {schema_type}"
+ else:
+ text = f"schema type: {schema_type}"
+
+ raise SchemaParseException(f"Default value <{default}> must match {text}")
+
+
+cdef _parse_schema(
+ schema,
+ namespace,
+ expand,
+ _write_hint,
+ names,
+ named_schemas,
+ default,
+ in_union=False,
+):
# union schemas
if isinstance(schema, list):
- return [
+ parsed_schemas = []
+ for index, s in enumerate(schema):
+ if index == 0:
+ parsed_schemas.append(
_parse_schema(
- s, namespace, expand, False, names, named_schemas
- ) for s in schema
- ]
+ s,
+ namespace,
+ expand,
+ False,
+ names,
+ named_schemas,
+ default,
+ in_union=True,
+ )
+ )
+ else:
+ parsed_schemas.append(
+ _parse_schema(
+ s, namespace, expand, False, names, named_schemas, NO_DEFAULT
+ )
+ )
+ return parsed_schemas
# string schemas; this could be either a named schema or a primitive type
elif not isinstance(schema, dict):
if schema in PRIMITIVES:
+ if default is not NO_DEFAULT:
+ if (
+ (schema == "null" and default is not None)
+ or (schema == "boolean" and not isinstance(default, bool))
+ or (schema == "string" and not isinstance(default, str))
+ or (schema == "bytes" and not isinstance(default, str))
+ or (schema == "double" and not isinstance(default, float))
+ or (schema == "float" and not isinstance(default, float))
+ or (schema == "int" and not isinstance(default, int))
+ or (schema == "long" and not isinstance(default, int))
+ ):
+ _raise_default_value_error(default, schema, in_union)
return schema
if "." not in schema and namespace:
@@ -206,7 +256,10 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
False,
names,
named_schemas,
+ NO_DEFAULT,
)
+ if default is not NO_DEFAULT and not isinstance(default, list):
+ _raise_default_value_error(default, schema_type, in_union)
elif schema_type == "map":
parsed_schema["values"] = _parse_schema(
@@ -216,7 +269,10 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
False,
names,
named_schemas,
+ NO_DEFAULT,
)
+ if default is not NO_DEFAULT and not isinstance(default, dict):
+ _raise_default_value_error(default, schema_type, in_union)
elif schema_type == "enum":
_, fullname = schema_name(schema, namespace)
@@ -226,6 +282,9 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
_validate_enum_symbols(schema)
+ if default is not NO_DEFAULT and not isinstance(default, str):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -237,6 +296,9 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
raise SchemaParseException(f"redefined named type: {fullname}")
names.add(fullname)
+ if default is not NO_DEFAULT and not isinstance(default, str):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -249,6 +311,9 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
raise SchemaParseException(f"redefined named type: {fullname}")
names.add(fullname)
+ if default is not NO_DEFAULT and not isinstance(default, dict):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
fields = []
@@ -274,6 +339,18 @@ cdef _parse_schema(schema, namespace, expand, _write_hint, names, named_schemas)
elif schema_type in PRIMITIVES:
parsed_schema["type"] = schema_type
+ if default is not NO_DEFAULT:
+ if (
+ (schema_type == "null" and default is not None)
+ or (schema_type == "boolean" and not isinstance(default, bool))
+ or (schema_type == "string" and not isinstance(default, str))
+ or (schema_type == "bytes" and not isinstance(default, str))
+ or (schema_type == "double" and not isinstance(default, float))
+ or (schema_type == "float" and not isinstance(default, float))
+ or (schema_type == "int" and not isinstance(default, int))
+ or (schema_type == "long" and not isinstance(default, int))
+ ):
+ _raise_default_value_error(default, schema_type, in_union)
else:
raise UnknownType(schema)
@@ -298,9 +375,11 @@ cdef parse_field(field, namespace, expand, names, named_schemas):
msg = f"aliases must be a list, not {aliases}"
raise SchemaParseException(msg)
+ default = field.get("default", NO_DEFAULT)
+
parsed_field["name"] = field["name"]
parsed_field["type"] = _parse_schema(
- field["type"], namespace, expand, False, names, named_schemas
+ field["type"], namespace, expand, False, names, named_schemas, default
)
return parsed_field

MODIFY
fastavro/_schema_py.py
fastavro/_schema_py.py
@@ -5,7 +5,7 @@ import math
from os import path
from copy import deepcopy
import re
-from typing import Tuple, Set, Optional, List
+from typing import Tuple, Set, Optional, List, Any
from .types import DictSchema, Schema, NamedSchemas
from .repository import (
@@ -28,6 +28,7 @@ from ._schema_common import (
)
SYMBOL_REGEX = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
+NO_DEFAULT = object()
def is_nullable_union(schema):
@@ -248,10 +249,14 @@ def parse_schema(
# Some old schemas might only have __fastavro_parsed and not
# __named_schemas since that came later. For these schemas, we need
# to re-parse the schema to handle named types
- return _parse_schema(schema, "", expand, _write_hint, set(), named_schemas)
+ return _parse_schema(
+ schema, "", expand, _write_hint, set(), named_schemas, NO_DEFAULT
+ )
if _force or expand:
- return _parse_schema(schema, "", expand, _write_hint, set(), named_schemas)
+ return _parse_schema(
+ schema, "", expand, _write_hint, set(), named_schemas, NO_DEFAULT
+ )
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
return schema
elif isinstance(schema, list):
@@ -264,7 +269,18 @@ def parse_schema(
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, NO_DEFAULT
+ )
+
+
+def _raise_default_value_error(default: Any, schema_type: str, in_union: bool):
+ if in_union:
+ text = f"first schema in union with type: {schema_type}"
+ else:
+ text = f"schema type: {schema_type}"
+
+ raise SchemaParseException(f"Default value <{default}> must match {text}")
def _parse_schema(
@@ -274,17 +290,50 @@ def _parse_schema(
_write_hint: bool,
names: Set[str],
named_schemas: NamedSchemas,
+ default: Any,
+ *,
+ in_union: bool = False,
) -> Schema:
# union schemas
if isinstance(schema, list):
- return [
- _parse_schema(s, namespace, expand, False, names, named_schemas)
- for s in schema
- ]
+ parsed_schemas = []
+ for index, s in enumerate(schema):
+ if index == 0:
+ parsed_schemas.append(
+ _parse_schema(
+ s,
+ namespace,
+ expand,
+ False,
+ names,
+ named_schemas,
+ default,
+ in_union=True,
+ )
+ )
+ else:
+ parsed_schemas.append(
+ _parse_schema(
+ s, namespace, expand, False, names, named_schemas, NO_DEFAULT
+ )
+ )
+ return parsed_schemas
# string schemas; this could be either a named schema or a primitive type
elif not isinstance(schema, dict):
if schema in PRIMITIVES:
+ if default is not NO_DEFAULT:
+ if (
+ (schema == "null" and default is not None)
+ or (schema == "boolean" and not isinstance(default, bool))
+ or (schema == "string" and not isinstance(default, str))
+ or (schema == "bytes" and not isinstance(default, str))
+ or (schema == "double" and not isinstance(default, float))
+ or (schema == "float" and not isinstance(default, float))
+ or (schema == "int" and not isinstance(default, int))
+ or (schema == "long" and not isinstance(default, int))
+ ):
+ _raise_default_value_error(default, schema, in_union)
return schema
if "." not in schema and namespace:
@@ -359,7 +408,10 @@ def _parse_schema(
False,
names,
named_schemas,
+ NO_DEFAULT,
)
+ if default is not NO_DEFAULT and not isinstance(default, list):
+ _raise_default_value_error(default, schema_type, in_union)
elif schema_type == "map":
parsed_schema["values"] = _parse_schema(
@@ -369,7 +421,10 @@ def _parse_schema(
False,
names,
named_schemas,
+ NO_DEFAULT,
)
+ if default is not NO_DEFAULT and not isinstance(default, dict):
+ _raise_default_value_error(default, schema_type, in_union)
elif schema_type == "enum":
_, fullname = schema_name(schema, namespace)
@@ -379,6 +434,9 @@ def _parse_schema(
_validate_enum_symbols(schema)
+ if default is not NO_DEFAULT and not isinstance(default, str):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -390,6 +448,9 @@ def _parse_schema(
raise SchemaParseException(f"redefined named type: {fullname}")
names.add(fullname)
+ if default is not NO_DEFAULT and not isinstance(default, str):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -402,6 +463,9 @@ def _parse_schema(
raise SchemaParseException(f"redefined named type: {fullname}")
names.add(fullname)
+ if default is not NO_DEFAULT and not isinstance(default, dict):
+ _raise_default_value_error(default, schema_type, in_union)
+
named_schemas[fullname] = parsed_schema
fields = []
@@ -425,6 +489,18 @@ def _parse_schema(
elif schema_type in PRIMITIVES:
parsed_schema["type"] = schema_type
+ if default is not NO_DEFAULT:
+ if (
+ (schema_type == "null" and default is not None)
+ or (schema_type == "boolean" and not isinstance(default, bool))
+ or (schema_type == "string" and not isinstance(default, str))
+ or (schema_type == "bytes" and not isinstance(default, str))
+ or (schema_type == "double" and not isinstance(default, float))
+ or (schema_type == "float" and not isinstance(default, float))
+ or (schema_type == "int" and not isinstance(default, int))
+ or (schema_type == "long" and not isinstance(default, int))
+ ):
+ _raise_default_value_error(default, schema_type, in_union)
else:
raise UnknownType(schema)
@@ -448,9 +524,11 @@ def parse_field(field, namespace, expand, names, named_schemas):
if not isinstance(aliases, list):
raise SchemaParseException(f"aliases must be a list, not {aliases}")
+ default = field.get("default", NO_DEFAULT)
+
parsed_field["name"] = field["name"]
parsed_field["type"] = _parse_schema(
- field["type"], namespace, expand, False, names, named_schemas
+ field["type"], namespace, expand, False, names, named_schemas, default
)
return parsed_field

MODIFY
tests/test_logical_types.py
tests/test_logical_types.py
@@ -562,7 +562,7 @@ def test_custom_logical_type_json_reader():
"fields": [
{
"name": "issue_json",
- "default": None,
+ "default": "{}",
"type": {"type": "string", "logicalType": "custom-json"},
},
],

MODIFY
tests/test_schema.py
tests/test_schema.py
@@ -1229,3 +1229,91 @@ def test_foobar():
records = list(fastavro.reader(bio))
assert records == [record, record]
+
+
+def test_default_matches_first_union_type():
+ """https://github.com/fastavro/fastavro/issues/649"""
+
+ schema = {
+ "type": "record",
+ "name": "test_default_matches_first_union_type",
+ "fields": [
+ {
+ "name": "first_union",
+ "type": ["string", "long"],
+ "default": 10,
+ }
+ ],
+ }
+
+ with pytest.raises(
+ SchemaParseException,
+ match="Default value <10> must match first schema in union",
+ ):
+ fastavro.parse_schema(schema)
+
+
+@pytest.mark.parametrize(
+ "field_type,default_value",
+ (
+ ("string", 0),
+ ({"type": "map", "values": "string"}, 0),
+ ({"type": "enum", "name": "somename", "symbols": ["FOO"]}, 0),
+ ({"type": "fixed", "name": "somename", "size": 4}, 0),
+ ({"type": "record", "name": "somename", "fields": []}, 0),
+ ({"type": "string"}, 0),
+ ),
+)
+def test_default_matches_type(field_type, default_value):
+ schema = {
+ "type": "record",
+ "name": "test_default_matches_type",
+ "fields": [
+ {
+ "name": "field",
+ "type": field_type,
+ "default": default_value,
+ }
+ ],
+ }
+
+ with pytest.raises(
+ SchemaParseException,
+ match=f"Default value <{default_value}> must match schema type",
+ ):
+ fastavro.parse_schema(schema)
+
+
+def test_correct_exception_text_for_default_value_test():
+ """https://github.com/fastavro/fastavro/issues/649"""
+
+ schema = {
+ "type": "record",
+ "name": "test_default_matches_first_union_type",
+ "fields": [
+ {
+ "name": "first_union",
+ "type": [
+ "null",
+ {
+ "type": "record",
+ "name": "internal",
+ "fields": [
+ {
+ "name": "internal_field",
+ "type": {"type": "array", "items": "string"},
+ "default": 10,
+ }
+ ],
+ },
+ ],
+ "default": None,
+ }
+ ],
+ }
+
+ with pytest.raises(
+ SchemaParseException,
+ match="Default value <10> must match schema type",
+ ):
+ fastavro.parse_schema(schema)

add default value checks when parsing the schema