def _add_schema()

in transcoder/output/avro/BaseAvroOutputManager.py [0:0]

• 11 lines of code
• 3 McCabe index (conditional complexity)

    def _add_schema(self, schema: DatacastSchema):
        # pylint: disable=duplicate-code
        _fields = self._get_field_list(schema.fields)
        if schema.name in self.schemas:
            del self.schemas[schema.name]
        if schema.name in self.writers:
            self.writers[schema.name].close()
            del self.writers[schema.name]

        schema_dict = {'type': 'record', 'namespace': 'sbeMessage', 'name': schema.name, 'fields': _fields}
        schema_json = json.dumps(schema_dict)
        self._save_schema(schema.name, schema_json)
        self.schemas[schema.name] = self._parse_schema(schema_dict)