def _create_data_set()

in source/forecast-shared/shared/quicksight_custom_resources/util/dataset.py [0:0]

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

    def _create_data_set(self, physical_table_map, logical_table_map):
        quicksight_client = get_quicksight_client()

        self._update_data_source_arn(physical_table_map)
        self._update_schema_arn(physical_table_map)
        self._update_table_name(physical_table_map)

        self.data_source.update()
        try:
            response = quicksight_client.create_data_set(
                AwsAccountId=self.aws_account_id,
                DataSetId=self.id,
                Name=self.name,
                Permissions=self._get_permissions(),
                PhysicalTableMap=physical_table_map,
                LogicalTableMap=logical_table_map,
                ImportMode="DIRECT_QUERY",  # TODO: support SPICE
            )
            logger.info(
                f"finished creating quicksight create_data_set id:{self.id}, "
                f"response:{response}"
            )
        except quicksight_client.exceptions.ResourceExistsException:
            logger.info(f"dataset for id:{self.id} already exists")
            response = quicksight_client.describe_data_set(
                AwsAccountId=self.aws_account_id, DataSetId=self.id
            )
            response = response["DataSet"]
        except quicksight_client.exceptions.InvalidParameterValueException as exc:
            logger.error(str(exc))
            raise QuickSightFailure()

        self.arn = response["Arn"]
        return response