def export()

in source/forecast-shared/shared/Predictor/predictor.py [0:0]
42 lines of code
5 McCabe index (conditional complexity)

    def export(self, dataset_file: DatasetFile) -> Export:
        """
        Export/ check on a predictor backtest import
        :param dataset_file: The dataset file last updated that generated this predictor
        :return: Status
        """
        if not self.arn:
            raise ValueError(
                "Predictor does not yet exist - cannot perform backtest export."
            )

        # Build the export arn
        export_arn = self.arn.replace(":predictor/", ":predictor-backtest-export-job/")
        if self.is_auto_predictor:
            export_name = f"export_{self._dataset_group.dataset_group_name}_auto_{self._latest_timestamp()}"
            export_arn = export_arn[:-27]
        else:
            export_name = f"export_{self._dataset_group.dataset_group_name}_{self._latest_timestamp()}"
        export_arn = export_arn + f"/{export_name}"

        past_export = Export()
        try:
            past_status = self.cli.describe_predictor_backtest_export_job(
                PredictorBacktestExportJobArn=export_arn
            )
            past_export.status = Status[past_status.get("Status")]
        except self.cli.exceptions.ResourceInUseException as excinfo:
            logger.debug(
                "Predictor backtest export %s is updating: %s"
                % (export_name, str(excinfo))
            )
        except self.cli.exceptions.ResourceNotFoundException:
            logger.info("Creating predictor backtest export %s" % export_name)
            self.cli.create_predictor_backtest_export_job(
                PredictorArn=self.arn,
                PredictorBacktestExportJobName=export_name,
                Destination={
                    "S3Config": {
                        "Path": f"s3://{dataset_file.bucket}/exports/{export_name}",
                        "RoleArn": environ.get("FORECAST_ROLE"),
                    }
                },
            )
            past_export.status = Status.CREATE_PENDING

        logger.debug(
            "Predictor backtest export status for %s is %s"
            % (export_name, str(past_export.status))
        )
        self.set_user_tags(resource_arn=export_arn)
        return past_export