AWS code-library documentation change
Summary
Added comprehensive Python code example demonstrating Neptune cluster lifecycle management including creation, status checks, start/stop operations, and deletion of clusters/instances/subnet groups
Security assessment
The changes add operational examples but don't address specific vulnerabilities. While security aspects like AccessDeniedException handling and IAM auth status display are present, these are standard AWS API interactions rather than new security documentation or vulnerability fixes.
Diff
diff --git a/code-library/latest/ug/neptune_example_neptune_Scenario_section.md b/code-library/latest/ug/neptune_example_neptune_Scenario_section.md index ef7444935..43dbc9261 100644 --- a//code-library/latest/ug/neptune_example_neptune_Scenario_section.md +++ b//code-library/latest/ug/neptune_example_neptune_Scenario_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: +The following code examples show how to: @@ -815,0 +816,692 @@ A wrapper class for Neptune SDK methods. +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). + + + import boto3 + import time + from botocore.exceptions import ClientError + + # Constants used in this scenario + POLL_INTERVAL_SECONDS = 10 + TIMEOUT_SECONDS = 1200 # 20 minutes + + def delete_db_cluster(neptune_client, cluster_id: str): + """ + Deletes a Neptune DB cluster and throws exceptions to the caller. + + Args: + neptune_client (boto3.client): The Neptune client object. + cluster_id (str): The ID of the Neptune DB cluster to be deleted. + + Raises: + ClientError: If the delete operation fails. + """ + request = { + 'DBClusterIdentifier': cluster_id, + 'SkipFinalSnapshot': True + } + + try: + print(f"Deleting DB Cluster: {cluster_id}") + neptune_client.delete_db_cluster(**request) + + except ClientError as err: + code = err.response["Error"]["Code"] + message = err.response["Error"]["Message"] + + if code == "DBClusterNotFoundFault": + print(f"Cluster '{cluster_id}' not found or already deleted.") + elif code == "AccessDeniedException": + print("Access denied. Please ensure you have the necessary permissions.") + else: + print(f"Couldn't delete DB cluster. {code}: {message}") + raise + + def format_elapsed_time(seconds: int) -> str: + mins, secs = divmod(seconds, 60) + hours, mins = divmod(mins, 60) + return f"{hours:02}:{mins:02}:{secs:02}" + + + def delete_db_instance(neptune_client, instance_id: str): + """ + Deletes a Neptune DB instance and waits for its deletion to complete. + Raises exception to be handled by calling code. + """ + print(f"Initiating deletion of DB Instance: {instance_id}") + try: + neptune_client.delete_db_instance( + DBInstanceIdentifier=instance_id, + SkipFinalSnapshot=True + ) + + print(f"Waiting for DB Instance '{instance_id}' to be deleted...") + waiter = neptune_client.get_waiter('db_instance_deleted') + waiter.wait( + DBInstanceIdentifier=instance_id, + WaiterConfig={ + 'Delay': 30, + 'MaxAttempts': 40 + } + ) + + print(f"DB Instance '{instance_id}' successfully deleted.") + + except ClientError as err: + code = err.response["Error"]["Code"] + message = err.response["Error"]["Message"] + + if code == "DBInstanceNotFoundFault": + print(f"Instance '{instance_id}' not found or already deleted.") + elif code == "AccessDeniedException": + print("Access denied. Please ensure you have the necessary permissions.") + else: + print(f"Couldn't delete DB instance. {code}: {message}") + raise + + def delete_db_subnet_group(neptune_client, subnet_group_name): + """ + Deletes a Neptune DB subnet group synchronously using Boto3. + + Args: + neptune_client (boto3.client): The Neptune client. + subnet_group_name (str): The name of the DB subnet group to delete. + + Raises: + ClientError: If the delete operation fails. + """ + delete_group_request = { + 'DBSubnetGroupName': subnet_group_name + } + + try: + neptune_client.delete_db_subnet_group(**delete_group_request) + print(f"️ Deleting Subnet Group: {subnet_group_name}") + + except ClientError as err: + code = err.response["Error"]["Code"] + message = err.response["Error"]["Message"] + + if code == "DBSubnetGroupNotFoundFault": + print(f"Subnet group '{subnet_group_name}' not found or already deleted.") + elif code == "AccessDeniedException": + print("Access denied. Please ensure you have the necessary permissions.") + else: + print(f"Couldn't delete subnet group. {code}: {message}") + raise + + def wait_for_cluster_status( + neptune_client, + cluster_id: str, + desired_status: str, + timeout_seconds: int = TIMEOUT_SECONDS, + poll_interval_seconds: int = POLL_INTERVAL_SECONDS + ): + """ + Waits for a Neptune DB cluster to reach a desired status. + + Args: + neptune_client (boto3.client): The Amazon Neptune client. + cluster_id (str): The identifier of the Neptune DB cluster. + desired_status (str): The target status (e.g., "available", "stopped"). + timeout_seconds (int): Max time to wait in seconds (default: 1200). + poll_interval_seconds (int): Polling interval in seconds (default: 10). + + Raises: + RuntimeError: If the desired status is not reached before timeout. + """ + print(f"Waiting for cluster '{cluster_id}' to reach status '{desired_status}'...") + start_time = time.time() + + while True: + # Prepare request object + describe_cluster_request = { + 'DBClusterIdentifier': cluster_id + } + + # Call the Neptune API + response = neptune_client.describe_db_clusters(**describe_cluster_request) + clusters = response.get('DBClusters', []) + current_status = clusters[0].get('Status') if clusters else None + elapsed_seconds = int(time.time() - start_time) + + status_str = current_status if current_status else "Unknown" + print( + f"\r Elapsed: {format_elapsed_time(elapsed_seconds):<20} Cluster status: {status_str:<20}", + end="", flush=True + ) + + if current_status and current_status.lower() == desired_status.lower(): + print( + f"\nNeptune cluster reached desired status '{desired_status}' after {format_elapsed_time(elapsed_seconds)}." + ) + return + + if elapsed_seconds > timeout_seconds: + raise RuntimeError(f"Timeout waiting for Neptune cluster to reach status: {desired_status}") + + time.sleep(poll_interval_seconds) + + + 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)