Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_enum(self):
test_schema = schema.EnumSchema('test_enum', None, ['A', 'B'], schema.Names())
self.assertEquals(self.converter.to_json_object('A', test_schema), 'A')
self.assertEquals(self.converter.from_json_object('B', test_schema), 'B')
def getContainerKeySchemaES(schema_files_location):
# Get the container event key schema
known_schemas = avro.schema.Names()
container_event_key_schema = LoadAvsc(schema_files_location + "/container_event_key_ES.avsc", known_schemas)
return container_event_key_schema
def to_json(self, names=None):
if names is None:
names = schema.Names()
to_dump = {}
to_dump['request'] = self.request.to_json(names)
to_dump['response'] = self.response.to_json(names)
if self.errors:
to_dump['errors'] = self.errors.to_json(names)
return to_dump
fail_msg = 'The name property must be a string.'
raise ProtocolParseException(fail_msg)
elif namespace is not None and not isinstance(namespace, basestring):
fail_msg = 'The namespace property must be a string.'
raise ProtocolParseException(fail_msg)
elif types is not None and not isinstance(types, list):
fail_msg = 'The types property must be a list.'
raise ProtocolParseException(fail_msg)
elif (messages is not None and
not(hasattr(messages, 'get') and callable(messages.get))):
fail_msg = 'The messages property must be a JSON object.'
raise ProtocolParseException(fail_msg)
self._props = {}
self.set_prop('name', name)
type_names = schema.Names()
if namespace is not None:
self.set_prop('namespace', namespace)
type_names.default_namespace = namespace
if types is not None:
self.set_prop('types', self._parse_types(types, type_names))
if messages is not None:
self.set_prop('messages', self._parse_messages(messages, type_names))
self._md5 = md5(str(self)).digest()
def _get_schema_names(self):
return (self._schema_names
if self.track_created_names
else schema.Names())
def reset(self):
self._schema_names = schema.Names()
def serialize(schema, data):
avro_schema = SchemaFromJSONData(schema, avro.schema.Names())
serializer = AvroJsonSerializer(avro_schema)
return serializer.to_json(data)
"""
Generate file containing concrete classes for RecordSchemas in given avro schema json
:param str schema_json: JSON representing avro schema
:param list[str] custom_imports: Add additional import modules
:param str avro_json_converter: AvroJsonConverter type to use for default values
:return Dict[str, str]:
"""
if avro_json_converter is None:
avro_json_converter = 'avrojson.AvroJsonConverter'
if '(' not in avro_json_converter:
avro_json_converter += '(use_logical_types=%s, schema_types=__SCHEMA_TYPES)' % use_logical_types
custom_imports = custom_imports or []
names = schema.Names()
make_avsc_object(json.loads(schema_json), names)
names = [k for k in six.iteritems(names.names) if isinstance(k[1], (schema.RecordSchema, schema.EnumSchema))]
names = sorted(names, key=lambda x: x[0])
main_out = StringIO()
writer = TabbedWriter(main_out)
write_preamble(writer, use_logical_types, custom_imports)
write_schema_preamble(writer)
write_get_schema(writer)
write_populate_schemas(writer)
writer.write('\n\n\nclass SchemaClasses(object):')
writer.tab()
writer.write('\n\n')
def to_json(self):
to_dump = {}
to_dump['protocol'] = self.name
names = schema.Names(default_namespace=self.namespace)
if self.namespace:
to_dump['namespace'] = self.namespace
if self.types:
to_dump['types'] = [ t.to_json(names) for t in self.types ]
if self.messages:
messages_dict = {}
for name, body in self.message_map.items():
messages_dict[name] = body.to_json(names)
to_dump['messages'] = messages_dict
return to_dump
def to_json(self):
to_dump = {}
to_dump['protocol'] = self.name
names = avro.schema.Names(default_namespace=self.namespace)
if self.namespace:
to_dump['namespace'] = self.namespace
if self.types:
to_dump['types'] = [ t.to_json(names) for t in self.types ]
if self.messages:
messages_dict = {}
for name, body in self.messages.iteritems():
messages_dict[name] = body.to_json(names)
to_dump['messages'] = messages_dict
return to_dump