Use DescribeDBClusters with an AWS SDK - AWS SDK Code Examples

There are more AWS SDK examples available in the AWS Doc SDK Examples GitHub repo.

Use DescribeDBClusters with an AWS SDK

The following code examples show how to use DescribeDBClusters.

Java
SDK for Java 2.x
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

/** * Asynchronously describes the specified Amazon RDS DB cluster. * * @param clusterId the identifier of the DB cluster to describe * @return a {@link CompletableFuture} that completes when the operation is done, or throws a {@link RuntimeException} * if an error occurs */ public CompletableFuture<Void> describeDBClustersAsync(String clusterId) { DescribeDbClustersRequest request = DescribeDbClustersRequest.builder() .dbClusterIdentifier(clusterId) .build(); return getAsyncClient().describeDBClusters(request) .thenAccept(response -> { for (DBCluster cluster : response.dbClusters()) { logger.info("Cluster Identifier: " + cluster.dbClusterIdentifier()); logger.info("Status: " + cluster.status()); logger.info("Engine: " + cluster.engine()); logger.info("Engine Version: " + cluster.engineVersion()); logger.info("Endpoint: " + cluster.endpoint()); logger.info("Reader Endpoint: " + cluster.readerEndpoint()); logger.info("Availability Zones: " + cluster.availabilityZones()); logger.info("Subnet Group: " + cluster.dbSubnetGroup()); logger.info("VPC Security Groups:"); cluster.vpcSecurityGroups().forEach(vpcGroup -> logger.info(" - " + vpcGroup.vpcSecurityGroupId())); logger.info("Storage Encrypted: " + cluster.storageEncrypted()); logger.info("IAM DB Auth Enabled: " + cluster.iamDatabaseAuthenticationEnabled()); logger.info("Backup Retention Period: " + cluster.backupRetentionPeriod() + " days"); logger.info("Preferred Backup Window: " + cluster.preferredBackupWindow()); logger.info("Preferred Maintenance Window: " + cluster.preferredMaintenanceWindow()); logger.info("------"); } }) .exceptionally(ex -> { Throwable cause = ex.getCause() != null ? ex.getCause() : ex; if (cause instanceof ResourceNotFoundException) { throw (ResourceNotFoundException) cause; } throw new RuntimeException("Failed to describe the DB cluster: " + cause.getMessage(), cause); }); }
Python
SDK for Python (Boto3)
Note

There's more on GitHub. Find the complete example and learn how to set up and run in the AWS Code Examples Repository.

def describe_db_clusters(neptune_client, cluster_id: str): """ Describes details of a Neptune DB cluster, paginating if needed. Args: neptune_client (boto3.client): The Neptune client. cluster_id (str): The ID of the cluster to describe. Raises: ClientError: If there's an AWS API error (e.g., cluster not found). """ paginator = neptune_client.get_paginator('describe_db_clusters') try: pages = paginator.paginate(DBClusterIdentifier=cluster_id) found = False for page in pages: for cluster in page.get('DBClusters', []): found = True print(f"Cluster Identifier: {cluster.get('DBClusterIdentifier')}") print(f"Status: {cluster.get('Status')}") print(f"Engine: {cluster.get('Engine')}") print(f"Engine Version: {cluster.get('EngineVersion')}") print(f"Endpoint: {cluster.get('Endpoint')}") print(f"Reader Endpoint: {cluster.get('ReaderEndpoint')}") print(f"Availability Zones: {cluster.get('AvailabilityZones')}") print(f"Subnet Group: {cluster.get('DBSubnetGroup')}") print("VPC Security Groups:") for vpc_group in cluster.get('VpcSecurityGroups', []): print(f" - {vpc_group.get('VpcSecurityGroupId')}") print(f"Storage Encrypted: {cluster.get('StorageEncrypted')}") print(f"IAM Auth Enabled: {cluster.get('IAMDatabaseAuthenticationEnabled')}") print(f"Backup Retention Period: {cluster.get('BackupRetentionPeriod')} days") print(f"Preferred Backup Window: {cluster.get('PreferredBackupWindow')}") print(f"Preferred Maintenance Window: {cluster.get('PreferredMaintenanceWindow')}") print("------") if not found: # Treat empty response as cluster not found raise ClientError( {"Error": {"Code": "DBClusterNotFound", "Message": f"No cluster found with ID '{cluster_id}'"}}, "DescribeDBClusters" ) except ClientError as err: code = err.response["Error"]["Code"] message = err.response["Error"]["Message"] if code == "AccessDeniedException": print("Access denied. Please ensure you have the necessary permissions.") elif code == "DBClusterNotFound": print(f"Cluster '{cluster_id}' not found. Please verify the cluster ID.") else: print(f"Couldn't describe DB cluster. Here's why: {code}: {message}") raise