AWS code-library documentation change
Summary
Added Python (Boto3) code example for StopDBCluster with error handling and status polling
Security assessment
Similar to StartDBCluster changes, this adds example code with standard error handling but no specific security vulnerability mitigation. The AccessDeniedException check is part of normal permission handling.
Diff
diff --git a/code-library/latest/ug/neptune_example_neptune_StopDBCluster_section.md b/code-library/latest/ug/neptune_example_neptune_StopDBCluster_section.md index 9d4bd8dfc..8dec769e7 100644 --- a//code-library/latest/ug/neptune_example_neptune_StopDBCluster_section.md +++ b//code-library/latest/ug/neptune_example_neptune_StopDBCluster_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 `StopDBCluster`. +The following code examples show how to use `StopDBCluster`. @@ -55,0 +56,77 @@ 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 stop_db_cluster(neptune_client, cluster_identifier: str): + """ + Stops an Amazon Neptune DB cluster and waits until it's fully stopped. + + Args: + neptune_client (boto3.client): The Neptune client. + cluster_identifier (str): The DB cluster identifier. + + Raises: + ClientError: For AWS API errors (e.g., resource not found). + RuntimeError: If the cluster doesn't stop within the timeout. + """ + try: + neptune_client.stop_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 stop 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() == 'stopped': + print(f"\nCluster '{cluster_identifier}' is now stopped.") + return + + if elapsed > TIMEOUT_SECONDS: + raise RuntimeError(f"Timeout waiting for cluster '{cluster_identifier}' to stop.") + + time.sleep(POLL_INTERVAL_SECONDS) + + + + + + * For API details, see [StopDBCluster](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/StopDBCluster) in _AWS SDK for Python (Boto3) API Reference_. + + + +