Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _parse(self, schema):
record_type = extract_record_type(schema)
if record_type == 'record':
production = []
production.append(RecordStart())
for field in schema["fields"]:
production.insert(0, FieldStart(field["name"]))
production.insert(0, self._parse(field["type"]))
production.insert(0, FieldEnd())
production.insert(0, RecordEnd())
seq = Sequence(*production)
return seq
elif record_type == 'union':
symbols = []
labels = []
for candidate_schema in schema:
symbols.append(self._parse(candidate_schema))
if isinstance(candidate_schema, dict):
labels.append(
candidate_schema.get(
"name",
candidate_schema.get("type")
)
)
else:
labels.append(candidate_schema)
self._parse(schema["values"]),
MapKeyMarker(),
String(),
)
return Sequence(repeat, MapStart())
elif record_type == "array":
repeat = Repeater(
ArrayEnd(),
ItemEnd(),
self._parse(schema["items"]),
)
return Sequence(repeat, ArrayStart())
elif record_type == "enum":
return Sequence(EnumLabels(schema["symbols"]), Enum())
elif record_type == "null":
return Null()
elif record_type == "boolean":
return Boolean()
elif record_type == "string":
return String()
elif record_type == "bytes":
return Bytes()
elif record_type == "int":
return Int()
elif record_type == "long":
return Long()
elif record_type == "float":
return Float()
elif record_type == "double":
)
)
else:
labels.append(candidate_schema)
return Sequence(Alternative(symbols, labels), Union())
elif record_type == "map":
repeat = Repeater(
MapEnd(),
# ItemEnd(), # TODO: Maybe need this?
self._parse(schema["values"]),
MapKeyMarker(),
String(),
)
return Sequence(repeat, MapStart())
elif record_type == "array":
repeat = Repeater(
ArrayEnd(),
ItemEnd(),
self._parse(schema["items"]),
)
return Sequence(repeat, ArrayStart())
elif record_type == "enum":
return Sequence(EnumLabels(schema["symbols"]), Enum())
elif record_type == "null":
return Null()
elif record_type == "boolean":
return Boolean()
repeat = Repeater(
MapEnd(),
# ItemEnd(), # TODO: Maybe need this?
self._parse(schema["values"]),
MapKeyMarker(),
String(),
)
return Sequence(repeat, MapStart())
elif record_type == "array":
repeat = Repeater(
ArrayEnd(),
ItemEnd(),
self._parse(schema["items"]),
)
return Sequence(repeat, ArrayStart())
elif record_type == "enum":
return Sequence(EnumLabels(schema["symbols"]), Enum())
elif record_type == "null":
return Null()
elif record_type == "boolean":
return Boolean()
elif record_type == "string":
return String()
elif record_type == "bytes":
return Bytes()
elif record_type == "int":
return Int()
elif record_type == "long":
return Long()