Moving QuickSight assets such as dashboards, analyses, and datasets from development to production can often feel complex and prone to manual errors. However, leveraging a custom Python script can automate the entire process, ensuring a seamless transition with minimal human intervention. In this guide, I’ll show you how to set up and use a Python script to export, update, and import your QuickSight assets while switching data sources and adjusting asset names for production use.
Overview
I’ll share how I designed a custom Python script to move QuickSight assets between environments, specifically from development (dev) to production (prod). My use case involved using the same AWS account but updating the data source for datasets while migrating dashboards and analyses without issues. Here’s a breakdown of the key elements:
- I used folders to manage user access for each dashboard.
- The goal was to update the dataset’s data source when moving from dev to prod while maintaining asset structure and content.
- Asset names in dev followed a convention with _dev, and I needed to remove _dev during the migration to production.
This blog will walk you through the process and explain how the script automates key steps like name changes, data source updates, and more. You’ll learn how to adapt this process to your own workflows for smooth QuickSight asset management.
The Initial Setup
# Input: JSON object containing dashboard names and corresponding ARNs
DASHBOARD_ARNS = {
"DASHBOARD_NAME": "arn:aws:quicksight:us-east-1:ACCOUNT_ID:dashboard/dashboard_id",
# Add other dashboards as needed
}
ACCOUNT_ID = "ACCOUNT_ID"
AWS_REGION = "us-east-1"
ASSETS = ["dataset", "dashboard", "analysis"]
ZIP_FOLDER = "zipfile/"
NEW_DATASOURCE_ARN = "arn:aws:quicksight:us-east-1:ACCOUNT_ID:datasource/DATASOURCE_ID"
FOLDER_ARN = "arn:aws:quicksight:us-east-1:ACCOUNT_ID:folder/FOLDER_ID"
SYSTEM = "prod"
NEW_ASSET_ARNS = []
client = boto3.client('quicksight', region_name=AWS_REGION)
This initial setup defines the necessary imports, sets the AWS region and account ID, and prepares the asset migration process by defining the ARNs of the dashboards.
Exporting the Assets
def start_export(dashboard_arn, export_id):
response = client.start_asset_bundle_export_job(
ResourceArns=[dashboard_arn],
ExportFormat="QUICKSIGHT_JSON",
IncludeAllDependencies=True,
AssetBundleExportJobId=export_id,
AwsAccountId=ACCOUNT_ID
)
def check_export_status(export_id):
try:
export_status = "IN_PROGRESS"
while export_status == "IN_PROGRESS":
response = client.describe_asset_bundle_export_job(
AssetBundleExportJobId=export_id,
AwsAccountId=ACCOUNT_ID
)
print(f'Export status: {response["JobStatus"]}')
if response["JobStatus"] == "SUCCESSFUL":
print("Export job completed")
return response["DownloadUrl"]
elif response["JobStatus"] == "FAILED":
print("Export job failed")
return ""
else:
print("Export job not completed yet")
time.sleep(10)
except Exception as e:
print(f"Failed due to {e}")
def unzip_file():
print("Unzipping export file")
import zipfile
with zipfile.ZipFile("export.zip", "r") as zip_ref:
zip_ref.extractall(ZIP_FOLDER)
print("Unzip complete")This function initiates the export of a QuickSight dashboard using the Export job. IncludeAllDependencies=True ensures the export of not only the dashboard but also related datasets and analyses, thereby packaging everything required for production. After receiving a zip file, we unzip it locally to update the assets before import.
Updating Asset IDs
QuickSight’s import job does not automatically generate new IDs for assets. This function adds a -prod suffix to the asset IDs, ensuring the imported assets don’t overwrite existing ones in the same account.
def update_asset_ids_in_file():
print("Updating asset IDs in exported files")
for root, dirs, files in os.walk(ZIP_FOLDER):
for filename in files:
if any(asset in root for asset in ASSETS):
filepath = os.path.join(root, filename)
try:
with open(filepath, 'r') as file:
data = json.load(file)
if 'dataSetId' in data:
new_id = f"{data['dataSetId']}-prod"
data['dataSetId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:dataset/{new_id}")
if 'dashboardId' in data:
new_id = f"{data['dashboardId']}-prod"
data['dashboardId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:dashboard/{new_id}")
if 'analysisId' in data:
new_id = f"{data['analysisId']}-prod"
data['analysisId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:analyses/{new_id}")
if 'logicalTableMap' in data:
for table_id, table in data['logicalTableMap'].items():
if 'source' in table and 'dataSetArn' in table['source']:
old_arn = table['source']['dataSetArn']
if not old_arn.endswith('-prod'):
table['source']['dataSetArn'] = old_arn + '-prod'
print(f"Updated logicalTableMap dataset ARN: {old_arn} -> {table['source']['dataSetArn']}")
with open(filepath, 'w') as file:
json.dump(data, file, indent=4)
print(f"Updated asset IDs in {filename}")
except Exception as e:
print(f"Error updating {filename}: {e}")
This step is crucial to avoid unintended updates to your dev assets.
Renaming and Updating Assets
In this specific case, the naming convention and same-account constraints required some asset renaming.
def update_asset_names():
dev_assets = ["dashboard", "analysis", "dataset"]
print("Updating asset names in exported files")
for root, dirs, files in os.walk(ZIP_FOLDER):
for filename in files:
filepath = os.path.join(root, filename)
try:
with open(filepath, 'r') as file:
data = json.load(file)
if 'name' in data and data['resourceType'] in dev_assets:
data['name'] = data['name'][:-4] # Remove "dev" or "_dev"
with open(filepath, 'w') as file:
json.dump(data, file, indent=4)
print(f"Updated Name in {filename}")
except Exception as e:
print(f"Error updating {filename}: {e}")
def update_asset_ids_in_file():
print("Updating asset IDs in exported files")
for root, dirs, files in os.walk(ZIP_FOLDER):
for filename in files:
if any(asset in root for asset in ASSETS):
filepath = os.path.join(root, filename)
try:
with open(filepath, 'r') as file:
data = json.load(file)
if 'dataSetId' in data:
new_id = f"{data['dataSetId']}-prod"
data['dataSetId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:dataset/{new_id}")
if 'dashboardId' in data:
new_id = f"{data['dashboardId']}-prod"
data['dashboardId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:dashboard/{new_id}")
if 'analysisId' in data:
new_id = f"{data['analysisId']}-prod"
data['analysisId'] = new_id
NEW_ASSET_ARNS.append(f"arn:aws:quicksight:{AWS_REGION}:{ACCOUNT_ID}:analyses/{new_id}")
if 'logicalTableMap' in data:
for table_id, table in data['logicalTableMap'].items():
if 'source' in table and 'dataSetArn' in table['source']:
old_arn = table['source']['dataSetArn']
if not old_arn.endswith('-prod'):
table['source']['dataSetArn'] = old_arn + '-prod'
print(f"Updated logicalTableMap dataset ARN: {old_arn} -> {table['source']['dataSetArn']}")
with open(filepath, 'w') as file:
json.dump(data, file, indent=4)
print(f"Updated asset IDs in {filename}")
except Exception as e:
print(f"Error updating {filename}: {e}")Assets are renamed by removing the _dev suffix to mark them as production-ready. Additionally, the data source ARN is updated to point to the production database, ensuring the assets use the correct source of data in the new environment.
Additionally, we updated the dataset ID to include the prod suffix in all the assets.
Updating the Datasource
def replace_datasource_in_file():
print("Replacing DataSourceArn in exported dataset files")
for root, dirs, files in os.walk(ZIP_FOLDER):
for filename in files:
if "dataset" in root:
filepath = os.path.join(root, filename)
try:
with open(filepath, 'r') as file:
data = json.load(file)
if 'physicalTableMap' in data:
for table in data['physicalTableMap'].values():
if 'customSql' in table and 'dataSourceArn' in table['customSql']:
table['customSql']['dataSourceArn'] = NEW_DATASOURCE_ARN
elif 'relationalTable' in table and 'dataSourceArn' in table['relationalTable']:
table['relationalTable']['dataSourceArn'] = NEW_DATASOURCE_ARN
with open(filepath, 'w') as file:
json.dump(data, file, indent=4)
print(f"Updated DataSourceArn in {filename}")
except Exception as e:
print(f"Error updating {filename}: {e}")
Here, the datasource arn is updated to the production datasource arn.
Importing the Assets to Production
def start_import(export_id):
with open('processed.zip', 'rb') as file:
response = client.start_asset_bundle_import_job(
AwsAccountId=ACCOUNT_ID,
AssetBundleImportJobId=export_id,
AssetBundleImportSource={'Body': file.read()},
FailureAction='ROLLBACK',
)
return response
def create_folder_if_not_exists(dashboard_name, folder_arn):
folder_name = dashboard_name.replace("_", "-")
folder_id = f"{folder_name}-prod"
try:
response = client.create_folder(
AwsAccountId=ACCOUNT_ID,
FolderId=folder_id,
Name=dashboard_name,
ParentFolderArn=folder_arn
)
print(f"Folder '{dashboard_name}' created successfully.")
return folder_id
except client.exceptions.ResourceExistsException:
print(f"Folder '{dashboard_name}' already exists. Skipping creation.")
return folder_id
except Exception as e:
print(f"Error creating folder: {e}")
return None
def add_assets_to_folder(folder_id):
print(f"Adding new assets to folder {folder_id}")
for asset_arn in NEW_ASSET_ARNS:
asset_type = asset_arn.split(":")[-1].split("/")[0].upper()
if asset_type == "ANALYSES":
asset_type = "ANALYSIS"
print(f"Folder id: {folder_id}")
response = client.create_folder_membership(
AwsAccountId=ACCOUNT_ID,
FolderId=folder_id,
MemberId=asset_arn.split("/")[-1],
MemberType=asset_type
)
print(f"Added {asset_type} {asset_arn} to folder.")
def check_import_status(folder_id, export_id):
import_status = "IN_PROGRESS"
new_asset_arns = []
while import_status == "IN_PROGRESS":
response = client.describe_asset_bundle_import_job(
AssetBundleImportJobId=export_id,
AwsAccountId=ACCOUNT_ID
)
print(f'Import status: {response["JobStatus"]}')
if response["JobStatus"] == "SUCCESSFUL":
print("Import job completed")
# Add imported assets to the created folder
add_assets_to_folder(folder_id)
return import_status
elif "FAILED" in response["JobStatus"] or "CANCELLED" in response["JobStatus"] or "TIMED_OUT" in response["JobStatus"]:
print("Import job failed")
print([x['Message'] for x in response['Errors']])
return ""
else:
print("Import job not completed yet")
time.sleep(10)
The import jobs re-zipped the assets and imported them into production after the necessary updates. The function places the assets in a specific folder using Create Folder Membership to maintain user permissions.
If you’re moving assets to a different account, you can skip the asset ID updates and focus only on data source updates to ensure correct data connections in the new environment.
Conclusion
By automating the migration of QuickSight assets using this Python script, you can significantly streamline your workflow, reduce errors, and ensure consistency when moving assets from dev to prod. Whether you’re working within the same account or across accounts, this method makes it easier to manage dashboards, datasets, and analyses across environments.
