AWS code-library documentation change
Summary
Added Python (Boto3) code example for StartDBCluster with error handling and status polling
Security assessment
The change adds example code with error handling (including AccessDeniedException), but there's no evidence this addresses a specific security vulnerability. The error handling appears to be standard API usage guidance.
Diff
diff --git a/code-library/latest/ug/neptune_example_neptune_StartDBCluster_section.md b/code-library/latest/ug/neptune_example_neptune_StartDBCluster_section.md index 6af5ba961..f3516cad3 100644 --- a//code-library/latest/ug/neptune_example_neptune_StartDBCluster_section.md +++ b//code-library/latest/ug/neptune_example_neptune_StartDBCluster_section.md @@ -9 +9 @@ There are more AWS SDK examples available in the [AWS Doc SDK Examples](https:// -The following code example shows how to use `StartDBCluster`. +The following code examples show how to use `StartDBCluster`. @@ -54,0 +55,78 @@ There's more on GitHub. Find the complete example and learn how to set up and ru +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](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/neptune#code-examples). + + + def start_db_cluster(neptune_client, cluster_identifier: str): + """ + Starts an Amazon Neptune DB cluster and waits until it reaches 'available'. + + Args: + neptune_client (boto3.client): The Neptune client. + cluster_identifier (str): The DB cluster identifier. + + Raises: + ClientError: Propagates AWS API issues like resource not found. + RuntimeError: If cluster doesn't reach 'available' within timeout. + """ + try: + # Initial wait in case the cluster was just stopped + time.sleep(30) + neptune_client.start_db_cluster(DBClusterIdentifier=cluster_identifier) + 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.") + else: + print(f"Couldn't start DB cluster. Here's why: {code}: {message}") + raise + + start_time = time.time() + paginator = neptune_client.get_paginator('describe_db_clusters') + + while True: + try: + pages = paginator.paginate(DBClusterIdentifier=cluster_identifier) + clusters = [] + for page in pages: + clusters.extend(page.get('DBClusters', [])) + except ClientError as err: + code = err.response["Error"]["Code"] + message = err.response["Error"]["Message"] + + if code == "DBClusterNotFound": + print(f"Cluster '{cluster_identifier}' not found while polling. It may have been deleted.") + else: + print(f"Couldn't describe DB cluster. Here's why: {code}: {message}") + raise + + status = clusters[0].get('Status') if clusters else None + elapsed = time.time() - start_time + + print(f"\rElapsed: {int(elapsed)}s ā Cluster status: {status}", end="", flush=True) + + if status and status.lower() == 'available': + print(f"\nš Cluster '{cluster_identifier}' is available.") + return + + if elapsed > TIMEOUT_SECONDS: + raise RuntimeError(f"Timeout waiting for cluster '{cluster_identifier}' to become available.") + + time.sleep(POLL_INTERVAL_SECONDS) + + + + + * For API details, see [StartDBCluster](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/StartDBCluster) in _AWS SDK for Python (Boto3) API Reference_. + + + +