def create_data_source()

in cid/helpers/quicksight.py [0:0]

• 45 lines of code
• 6 McCabe index (conditional complexity)

    def create_data_source(self) -> bool:
        """Create a new data source"""
        logger.info('Creating Athena data source')
        params = {
            "AwsAccountId": self.account_id,
            "DataSourceId": "95aa6f18-abb4-436f-855f-182b199a961f",
            "Name": "Athena",
            "Type": "ATHENA",
            "DataSourceParameters": {
                "AthenaParameters": {
                    "WorkGroup": "primary"
                }
            },
            "Permissions": [
                {
                    "Principal": self.user.get('Arn'),
                    "Actions": [
                        "quicksight:UpdateDataSourcePermissions",
                        "quicksight:DescribeDataSource",
                        "quicksight:DescribeDataSourcePermissions",
                        "quicksight:PassDataSource",
                        "quicksight:UpdateDataSource",
                        "quicksight:DeleteDataSource"
                    ]
                }
            ]
        }
        try:
            logger.info(f'Creating data source {params}')
            create_status = self.client.create_data_source(**params)
            logger.debug(f'Data source creation result {create_status}')
            current_status = create_status['CreationStatus']
            logger.info(f'Data source creation status {current_status}')
            # Poll for the current status of query as long as its not finished
            while current_status in ['CREATION_IN_PROGRESS', 'UPDATE_IN_PROGRESS']:
                response = self.describe_data_source(create_status['DataSourceId'])
                current_status = response.get('Status')

            if (current_status != "CREATION_SUCCESSFUL"):
                failure_reason = response.get('Errors')
                logger.info(f'Data source creation failed with reason {failure_reason}')
                return False
            return True
        except self.client.exceptions.ResourceExistsException:
            logger.error('Data source already exists')
        except self.client.exceptions.AccessDeniedException:
            logger.info('Access denied creating data source')
        return False