def generate_workflows_body()

in workflows-generator/ComposerDagGenerator.py [0:0]

• 8 lines of code
• 1 McCabe index (conditional complexity)

    def generate_workflows_body(self):
        """method to generate Airflow body"""
        levels = self.process_levels(self.workflow_config)
        workflow_body = self.workflow_template.replace("<<LEVELS>>", "".join(levels))
        workflow_body = workflow_body.replace("<<LEVEL_DEPENDENCIES>>",
                                              self.get_level_dependency_string(self.workflow_config))
        workflow_body = workflow_body.replace("<<DAG_NAME>>", self.json_file_name)
        workflow_body = workflow_body.replace("<<STEPS_ARGS>>", self.process_steps_vars(self.workflow_config))
        return workflow_body