def apply_input_mappings()

in source/glue/jobs/forecast_etl.py [0:0]

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

    def apply_input_mappings(self):
        """
        Map all fields to those supported in consolidating
        :return: The DynamicFame
        """
        logger.info(
            "%s applying schema mappings"  # NOSONAR (python:S1192) - string for clarity
            % self.etl.name
        )

        fields = list(self.df.schema().field_map.keys())
        mappings = list()
        for field in fields:
            if field == self.etl.identifier:
                mappings.append((field, "string", "identifier", "string"))
            elif field == "timestamp" or field == "date":
                mappings.append((field, "string", "timestamp", "timestamp"))
            elif field == self.etl.target_field or field == "target_value":
                mappings.append((field, "string", "metric", "double"))
            elif QUANTILE_RE.match(field):
                mappings.append((field, "string", field, "double"))
            elif (
                field == "backtestwindow_start_time"
                or field == "backtestwindow_end_time"
            ):
                pass  # ignore - forecast does not allow overlapping backtest windows as of 2021-02-04
            else:
                self.output_dimensions.append(field)
                mappings.append((field, "string", field, "string"))

        return ApplyMapping.apply(
            frame=self.df,
            mappings=mappings,
            transformation_ctx="ApplyMapping",
        )