Amazon Quicksight- Automating asset deployment between environments using Python
Tags: automation, amazonquicksight, visualization, deployment
Published: 2023-12-28 21:09:03

Quicksight previously had offered variety of APIs to move assets between different accounts for example between dev, test and prod AWS accounts. To make things quicker and easier they have introduced new API's using a concept called Asset as Bundle. So all our Quicksight assets namely Datasets, Analysis & Dashboards and this also includes themes, Vpc configurations are packed together as a bundle and that makes it easy to move those assets between different accounts without having to move every asset separately which was the case previously.

you could use any programming language to automate the [Quicksight APIs](https://docs.aws.amazon.com/cli/latest/reference/quicksight/)


The new APIs allows you to interact with all the assets in a lift-and-shift manner for deployment across QuickSight accounts. These APIs can also be used for backing up assets and restoring them when needed. BIOps teams will benefit from this capability allowing them to automate Quicksight assets backup and deployment

The two types of APIs are:

1. Export APIS
2. Import APIS



 **Export APIS**-  This can be used to initiate, track and describe export jobs. When you initiate the export job it creates a bundle file as a zip file (.qs extension) which basically has all your quicksight assets. 

**StartAssetBundleExportJob** - Asynchronous API to export an asset bundle file.

**DescribeAssetBundleExportJob** -  Synchronous API to get the status of your export job. When successful, this API call response will have a presigned URL to fetch the asset bundle.

**ListAssetBundleExportJobs** -  Synchronous API to list past export jobs. The list will contain both finished and running jobs from the past 15 days.


 **Import APIS**-  his can be used to initiate, track and describe import jobs. These apis can be used on the target quicksight account.

StartAssetBundleImportJob - Asynchronous API to import an asset bundle file.

DescribeAssetBundleImportJob - Synchronous API to get the status of your import job.

ListAssetBundleImportJobs- Synchronous API to list past import jobs. The list will contain both finished and running jobs from the past 15 days.


**Steps for Export job:**

1. Use the StartAssetBundleExportJob API and export assets into a bundle file.
2. Use DescribeAssetBundleExportJob API to view the status and presigned URL that you will use to put it into a S3 bucket.
3. Place the bundle file in a S3 bucket.
4. Use ListAssetBundleExportJobs API to list what assets have been exported.

**Steps for Import job:**

1. Use the StartAssetBundleImportJob API and get the assets from S3 bucket overriding the source details.
2. Use DescribeAssetBundleImportJob API to view the status of the import.
3. Use ListAssetBundleImportJobsAPI to list what assets have been imported.


**Python code sample to automate assets deployments:**

import boto3
import re
import json
import time
import requests

SourceAccountID=<Source Account ID>
SourceDashboardName=<Dashboard Name>
SourceDashboardId=<Dashboard ID>

SourceAnalysisName=<Analysis Name>
SourceAnalysisId=<Analysis ID>
SourceRoleName='QuickSightFullAccess'
SourceRegion=<Source Account Region>
IncludeDependencies=True

Resource_Arns_Dashboard=<Resource ARN for Dashboard>
Resource_Arns_Analysis=<Resource ARN for Analysis>


account_id=<AWS Account ID>
analysis_id=<Analysis ID>
region=<Account Region>
dashboardId=<Dashboard ID>

client = boto3.client('quicksight', region)
client_s3 = boto3.client('s3', region)


# Call describe_analysis() to retrieve the analysis details

try:
    response=client.**start_asset_bundle_export_job**(
        AwsAccountId=SourceAccountID ,
        AssetBundleExportJobId= 'job-1',
        ResourceArns= Resource_Arns_Analysis, 
        ExportFormat='QUICKSIGHT_JSON',
        IncludeAllDependencies=IncludeDependencies
        
        )
    print(response)
    
    time.sleep(60)
    
    
    res=client.**describe_asset_bundle_export_job**(
        AwsAccountId=SourceAccountID,
        AssetBundleExportJobId='job-1'        
        
        )
    
    print(res)
    
    # save the exported asset bundle file into S3

    for key,value in res.items():
        if (key=='DownloadUrl'):
            download_url=value;
            print(download_url)
    
            
    r=requests.get(download_url, allow_redirects=True)
    open('job-1.zip','wb').write(r.content)
    
    
    #target_s3_client=targetsession.client('s3')
    
    bucket_name="assetbundle-" + SourceAccountID + "-" + SourceRegion
    

# uploading file to s3 bucket

    client_s3.upload_file(
        
        Filename='job-1.zip',
        Bucket="qsbackup",
        Key='Imported/job-1.zip' 
        
        )
    
    
    # import dashboard back to target account 
    
    res_import=client.**start_asset_bundle_import_job**(
        AwsAccountId=SourceAccountID,
        AssetBundleImportJobId='Test_Import',
        AssetBundleImportSource={"S3Uri":"s3://assetbundle-" +SourceAccountID+ "-" +SourceRegion+"/Imported/job-1.zip"},
        OverrideParameters={
            
             'Analyses': [
            {
                'AnalysisId': <Analysis ID>,
                'Name': 'NEWLY_IMPORTED_USING_API'
            },
            
            ]
             
            }
        
        )
    
    print(res_import)
    
    time.sleep(60)

    res_import_describe=client.**describe_asset_bundle_import_job**(
        AwsAccountId=SourceAccountID,
        AssetBundleImportJobId='Test_Import'        
        
        )
    
    print(res_import_describe)
    
    
    # update permissions
    
    res_update_permissions=client.**update_analysis_permissions**(
        AwsAccountId=SourceAccountID,
        AnalysisId=SourceAnalysisId, 
        GrantPermissions= [
            {
                
                'Principal':<principal ARN> ,
                'Actions': [
                    "quicksight:RestoreAnalysis",
                    "quicksight:UpdateAnalysisPermissions",
                    "quicksight:DeleteAnalysis",
                    "quicksight:QueryAnalysis",
                    "quicksight:DescribeAnalysisPermissions",
                    "quicksight:DescribeAnalysis",
                    "quicksight:UpdateAnalysis"
                    
                    
                    ]
                
                
                }
            
            
            
            
            ]
      
        
        )
    
    print(res_update_permissions)
    
    #analysis_name = response['Analysis']['Name']
    dashboard = response
    print(dashboard)
    
except TypeError as e:
    print('Caught TypeError:', e)    
    
print('End')