ensure that named schemas are not redefined
MODIFY
fastavro/_schema.pyx
fastavro/_schema.pyx
@@ -52,17 +52,21 @@ cpdef schema_name(schema, parent_ns):
def parse_schema(schema, _write_hint=True, _force=False):
if _force:
- return _parse_schema(schema, "", _write_hint)
+ return _parse_schema(schema, "", _write_hint, set())
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
return schema
else:
- return _parse_schema(schema, "", _write_hint)
+ return _parse_schema(schema, "", _write_hint, set())
-cdef _parse_schema(schema, namespace, _write_hint):
+cdef _parse_schema(schema, namespace, _write_hint, named_schemas):
# union schemas
if isinstance(schema, list):
- return [_parse_schema(s, namespace, False) for s in schema]
+ return [
+ _parse_schema(
+ s, namespace, False, named_schemas
+ ) for s in schema
+ ]
# string schemas; this could be either a named schema or a primitive type
elif not isinstance(schema, dict):
@@ -109,6 +113,7 @@ cdef _parse_schema(schema, namespace, _write_hint):
schema["items"],
namespace,
False,
+ named_schemas,
)
elif schema_type == "map":
@@ -116,10 +121,17 @@ cdef _parse_schema(schema, namespace, _write_hint):
schema["values"],
namespace,
False,
+ named_schemas,
)
elif schema_type == "enum":
_, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -127,6 +139,12 @@ cdef _parse_schema(schema, namespace, _write_hint):
elif schema_type == "fixed":
_, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -135,12 +153,18 @@ cdef _parse_schema(schema, namespace, _write_hint):
elif schema_type == "record" or schema_type == "error":
# records
namespace, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
fields = []
for field in schema.get('fields', []):
fields.append(
- parse_field(field, namespace)
+ parse_field(field, namespace, named_schemas)
)
parsed_schema["name"] = fullname
@@ -159,7 +183,7 @@ cdef _parse_schema(schema, namespace, _write_hint):
return parsed_schema
-cdef parse_field(field, namespace):
+cdef parse_field(field, namespace, named_schemas):
parsed_field = {
key: value
for key, value in iteritems(field)
@@ -177,7 +201,9 @@ cdef parse_field(field, namespace):
raise SchemaParseException(msg)
parsed_field["name"] = field["name"]
- parsed_field["type"] = _parse_schema(field["type"], namespace, False)
+ parsed_field["type"] = _parse_schema(
+ field["type"], namespace, False, named_schemas
+ )
return parsed_field

MODIFY
fastavro/_schema_py.py
fastavro/_schema_py.py
@@ -79,17 +79,21 @@ def parse_schema(schema, _write_hint=True, _force=False):
writer(out, parsed_schema, records)
"""
if _force:
- return _parse_schema(schema, "", _write_hint)
+ return _parse_schema(schema, "", _write_hint, set())
elif isinstance(schema, dict) and "__fastavro_parsed" in schema:
return schema
else:
- return _parse_schema(schema, "", _write_hint)
+ return _parse_schema(schema, "", _write_hint, set())
-def _parse_schema(schema, namespace, _write_hint):
+def _parse_schema(schema, namespace, _write_hint, named_schemas):
# union schemas
if isinstance(schema, list):
- return [_parse_schema(s, namespace, False) for s in schema]
+ return [
+ _parse_schema(
+ s, namespace, False, named_schemas
+ ) for s in schema
+ ]
# string schemas; this could be either a named schema or a primitive type
elif not isinstance(schema, dict):
@@ -136,6 +140,7 @@ def _parse_schema(schema, namespace, _write_hint):
schema["items"],
namespace,
False,
+ named_schemas,
)
elif schema_type == "map":
@@ -143,10 +148,17 @@ def _parse_schema(schema, namespace, _write_hint):
schema["values"],
namespace,
False,
+ named_schemas,
)
elif schema_type == "enum":
_, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -154,6 +166,12 @@ def _parse_schema(schema, namespace, _write_hint):
elif schema_type == "fixed":
_, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
parsed_schema["name"] = fullname
@@ -162,12 +180,18 @@ def _parse_schema(schema, namespace, _write_hint):
elif schema_type == "record" or schema_type == "error":
# records
namespace, fullname = schema_name(schema, namespace)
+ if fullname in named_schemas:
+ raise SchemaParseException(
+ "redefined named type: {}".format(fullname)
+ )
+ named_schemas.add(fullname)
+
SCHEMA_DEFS[fullname] = parsed_schema
fields = []
for field in schema.get('fields', []):
fields.append(
- parse_field(field, namespace)
+ parse_field(field, namespace, named_schemas)
)
parsed_schema["name"] = fullname
@@ -186,7 +210,7 @@ def _parse_schema(schema, namespace, _write_hint):
return parsed_schema
-def parse_field(field, namespace):
+def parse_field(field, namespace, named_schemas):
parsed_field = {
key: value
for key, value in iteritems(field)
@@ -204,7 +228,9 @@ def parse_field(field, namespace):
raise SchemaParseException(msg)
parsed_field["name"] = field["name"]
- parsed_field["type"] = _parse_schema(field["type"], namespace, False)
+ parsed_field["type"] = _parse_schema(
+ field["type"], namespace, False, named_schemas
+ )
return parsed_field

MODIFY
tests/test_fastavro.py
tests/test_fastavro.py
@@ -1449,11 +1449,9 @@ def test_passing_same_schema_to_reader():
'default': False,
}, {
'name': 'outcome',
- 'type': [{
- 'type': 'record',
- 'name': 'SomeMessage',
- 'fields': [],
- }, {
+ 'type': [
+ "SomeMessage",
+ {
'type': 'record',
'name': 'ErrorRecord',
'fields': [{
@@ -1461,7 +1459,8 @@ def test_passing_same_schema_to_reader():
'type': {'type': 'map', 'values': 'string'},
'doc': 'doc',
}]
- }]
+ }
+ ]
}]
}

MODIFY
tests/test_schema.py
tests/test_schema.py
@@ -149,3 +149,86 @@ def test_precision_is_an_int():
parse_schema(schema)
assert "decimal precision must be a postive integer" in str(exc)
+
+
+def test_named_type_cannot_be_redefined():
+ schema = {
+ "type": "record",
+ "namespace": "test.avro.training",
+ "name": "SomeMessage",
+ "fields": [{
+ "name": "is_error",
+ "type": "boolean",
+ "default": False,
+ }, {
+ "name": "outcome",
+ "type": [{
+ "type": "record",
+ "name": "SomeMessage",
+ "fields": [],
+ }, {
+ "type": "record",
+ "name": "ErrorRecord",
+ "fields": [{
+ "name": "errors",
+ "type": {"type": "map", "values": "string"},
+ "doc": "doc",
+ }],
+ }],
+ }],
+ }
+
+ with pytest.raises(SchemaParseException) as exc:
+ parse_schema(schema)
+
+ assert "redefined named type: test.avro.training.SomeMessage" in str(exc)
+
+ schema = {
+ "type": "record",
+ "name": "SomeMessage",
+ "fields": [{
+ "name": "field1",
+ "type": {
+ "type": "record",
+ "name": "ThisName",
+ "fields": [],
+ },
+ }, {
+ "name": "field2",
+ "type": {
+ "type": "enum",
+ "name": "ThisName",
+ "symbols": ["FOO", "BAR"],
+ },
+ }],
+ }
+
+ with pytest.raises(SchemaParseException) as exc:
+ parse_schema(schema)
+
+ assert "redefined named type: ThisName" in str(exc)
+
+ schema = {
+ "type": "record",
+ "name": "SomeMessage",
+ "fields": [{
+ "name": "field1",
+ "type": {
+ "type": "record",
+ "name": "ThatName",
+ "fields": [],
+ },
+ }, {
+ "name": "field2",
+ "type": {
+ "type": "fixed",
+ "name": "ThatName",
+ "size": 8,
+ },
+ }],
+ }
+
+ with pytest.raises(SchemaParseException) as exc:
+ parse_schema(schema)
+
+ assert "redefined named type: ThatName" in str(exc)

ensure that named schemas are not redefined