def _add_schema()

in transcoder/output/google_cloud/terraform/PubSubTerraformOutputManager.py [0:0]

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

    def _add_schema(self, schema: DatacastSchema):
        _fields = self._get_field_list(schema.fields)
        schema_dict = {'type': 'record', 'namespace': 'sbeMessage', 'name': schema.name, 'fields': _fields}
        schema_json = json.dumps(schema_dict)  # .replace('"', '\\"')
        schema_json = json.dumps(schema_json)

        content = ""
        if self.create_schema_enforcing_topics is True:
            content += f"""resource \"google_pubsub_schema\" \"{schema.name}\" {{
  name = \"{schema.name}\"
  type = \"AVRO\"
  definition = {schema_json}
}}

"""

            content += f"""resource \"google_pubsub_topic\" \"{schema.name}\" {{
  name = \"{schema.name}\"

  schema_settings {{
    schema = \"projects/{self.project_id}/schemas/{schema.name}\"
    encoding = \"{'BINARY' if self.is_binary_encoded is True else 'JSON'}\"
  }}
  
  labels = {{
    \"{GOOGLE_PACKAGED_SOLUTION_KEY}\" = \"{GOOGLE_PACKAGED_SOLUTION_VALUE}\"
  }}
  
  depends_on = [google_pubsub_schema.{schema.name}]
}}
"""

        self._save_schema(schema.name, content)