There are more AWS SDK examples available in the [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) GitHub repo. # Actions for Neptune using AWS SDKs The following code examples demonstrate how to perform individual Neptune actions with AWS SDKs. Each example includes a link to GitHub, where you can find instructions for setting up and running the code. These excerpts call the Neptune API and are code excerpts from larger programs that must be run in context. You can see actions in context in [Scenarios for Neptune using AWS SDKs](neptune_code_examples_scenarios.md). The following examples include only the most commonly used actions. For a complete list, see the [Amazon Neptune API Reference](https://docs.aws.amazon.com/neptune/latest/apiref/Welcome.html). **Topics** + [`CreateDBCluster`](neptune_example_neptune_CreateDBCluster_section.md) + [`CreateDBInstance`](neptune_example_neptune_CreateDBInstance_section.md) + [`CreateDBSubnetGroup`](neptune_example_neptune_CreateDBSubnetGroup_section.md) + [`CreateGraph`](neptune_example_neptune_CreateGraph_section.md) + [`DeleteDBCluster`](neptune_example_neptune_DeleteDBCluster_section.md) + [`DeleteDBInstance`](neptune_example_neptune_DeleteDBInstance_section.md) + [`DeleteDBSubnetGroup`](neptune_example_neptune_DeleteDBSubnetGroup_section.md) + [`DescribeDBClusters`](neptune_example_neptune_DescribeDBClusters_section.md) + [`DescribeDBInstances`](neptune_example_neptune_DescribeDBInstances_section.md) + [`ExecuteGremlinProfileQuery`](neptune_example_neptune_ExecuteGremlinProfileQuery_section.md) + [`ExecuteGremlinQuery`](neptune_example_neptune_ExecuteGremlinQuery_section.md) + [`ExecuteOpenCypherExplainQuery`](neptune_example_neptune_ExecuteOpenCypherExplainQuery_section.md) + [`ExecuteQuery`](neptune_example_neptune_ExecuteQuery_section.md) + [`StartDBCluster`](neptune_example_neptune_StartDBCluster_section.md) + [`StopDBCluster`](neptune_example_neptune_StopDBCluster_section.md) # Use `CreateDBCluster` with an AWS SDK The following code examples show how to use `CreateDBCluster`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Creates a new Amazon Neptune DB cluster asynchronously. * * @param dbName the name of the DB cluster to be created * @return a CompletableFuture that, when completed, provides the ID of the created DB cluster * @throws CompletionException if the operation fails for any reason, including if the request would exceed the maximum quota */ public CompletableFuture createDBClusterAsync(String dbName) { CreateDbClusterRequest request = CreateDbClusterRequest.builder() .dbClusterIdentifier(dbName) .engine("neptune") .deletionProtection(false) .backupRetentionPeriod(1) .build(); return getAsyncClient().createDBCluster(request) .whenComplete((response, exception) -> { if (exception != null) { Throwable cause = exception.getCause(); if (cause instanceof ServiceQuotaExceededException) { throw new CompletionException("The operation was denied because the request would exceed the maximum quota.", cause); } throw new CompletionException("Failed to create Neptune DB cluster: " + exception.getMessage(), exception); } }) .thenApply(response -> { String clusterId = response.dbCluster().dbClusterIdentifier(); logger.info("DB Cluster created: " + clusterId); return clusterId; }); } ``` + For API details, see [CreateDBCluster](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/CreateDBCluster) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 create_db_cluster(neptune_client, db_name: str) -> str: """ Creates a Neptune DB cluster and returns its identifier. Args: neptune_client (boto3.client): The Neptune client object. db_name (str): The desired cluster identifier. Returns: str: The DB cluster identifier. Raises: RuntimeError: For any failure or AWS error, with a user-friendly message. """ request = { 'DBClusterIdentifier': db_name, 'Engine': 'neptune', 'DeletionProtection': False, 'BackupRetentionPeriod': 1 } try: response = neptune_client.create_db_cluster(**request) cluster = response.get('DBCluster') or {} cluster_id = cluster.get('DBClusterIdentifier') if not cluster_id: raise RuntimeError("Cluster created but no ID returned.") print(f"DB Cluster created: {cluster_id}") return cluster_id except ClientError as e: code = e.response["Error"]["Code"] message = e.response["Error"]["Message"] if code in ("ServiceQuotaExceededException", "DBClusterQuotaExceededFault"): raise RuntimeError("You have exceeded the quota for Neptune DB clusters.") from e else: raise RuntimeError(f"AWS error [{code}]: {message}") from e except Exception as e: raise RuntimeError(f"Unexpected error creating DB cluster '{db_name}': {e}") from e ``` + For API details, see [CreateDBCluster](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/CreateDBCluster) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `CreateDBInstance` with an AWS SDK The following code examples show how to use `CreateDBInstance`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Creates a new Amazon Neptune DB instance asynchronously. * * @param dbInstanceId the identifier for the new DB instance * @param dbClusterId the identifier for the DB cluster that the new instance will be a part of * @return a {@link CompletableFuture} that completes with the identifier of the newly created DB instance * @throws CompletionException if the operation fails, with a cause of either: * - {@link ServiceQuotaExceededException} if the request would exceed the maximum quota, or * - a general exception with the failure message */ public CompletableFuture createDBInstanceAsync(String dbInstanceId, String dbClusterId) { CreateDbInstanceRequest request = CreateDbInstanceRequest.builder() .dbInstanceIdentifier(dbInstanceId) .dbInstanceClass("db.r5.large") .engine("neptune") .dbClusterIdentifier(dbClusterId) .build(); return getAsyncClient().createDBInstance(request) .whenComplete((response, exception) -> { if (exception != null) { Throwable cause = exception.getCause(); if (cause instanceof ServiceQuotaExceededException) { throw new CompletionException("The operation was denied because the request would exceed the maximum quota.", cause); } throw new CompletionException("Failed to create Neptune DB instance: " + exception.getMessage(), exception); } }) .thenApply(response -> { String instanceId = response.dbInstance().dbInstanceIdentifier(); logger.info("Created Neptune DB Instance: " + instanceId); return instanceId; }); } ``` + For API details, see [CreateDBInstance](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/CreateDBInstance) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 create_db_instance(neptune_client, db_instance_id: str, db_cluster_id: str) -> str: try: request = { 'DBInstanceIdentifier': db_instance_id, 'DBInstanceClass': 'db.r5.large', 'Engine': 'neptune', 'DBClusterIdentifier': db_cluster_id } print(f"Creating Neptune DB Instance: {db_instance_id}") response = neptune_client.create_db_instance(**request) instance = response.get('DBInstance') if not instance or 'DBInstanceIdentifier' not in instance: raise RuntimeError("Instance creation succeeded but no ID returned.") print(f"Waiting for DB Instance '{db_instance_id}' to become available...") waiter = neptune_client.get_waiter('db_instance_available') waiter.wait( DBInstanceIdentifier=db_instance_id, WaiterConfig={'Delay': 30, 'MaxAttempts': 40} ) print(f"DB Instance '{db_instance_id}' is now available.") return instance['DBInstanceIdentifier'] 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 create DB instance. Here's why: {code}: {message}") raise except Exception as e: print(f"Unexpected error creating DB instance '{db_instance_id}': {e}") raise RuntimeError(f"Unexpected error creating DB instance '{db_instance_id}': {e}") from e ``` + For API details, see [CreateDBInstance](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/CreateDBInstance) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `CreateDBSubnetGroup` with an AWS SDK The following code examples show how to use `CreateDBSubnetGroup`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Creates a new DB subnet group asynchronously. * * @param groupName the name of the subnet group to create * @return a CompletableFuture that, when completed, returns the Amazon Resource Name (ARN) of the created subnet group * @throws CompletionException if the operation fails, with a cause that may be a ServiceQuotaExceededException if the request would exceed the maximum quota */ public CompletableFuture createSubnetGroupAsync(String groupName) { // Get the Amazon Virtual Private Cloud (VPC) where the Neptune cluster and resources will be created String vpcId = getDefaultVpcId(); logger.info("VPC is : " + vpcId); List subnetList = getSubnetIds(vpcId); for (String subnetId : subnetList) { System.out.println("Subnet group:" +subnetId); } CreateDbSubnetGroupRequest request = CreateDbSubnetGroupRequest.builder() .dbSubnetGroupName(groupName) .dbSubnetGroupDescription("Subnet group for Neptune cluster") .subnetIds(subnetList) .build(); return getAsyncClient().createDBSubnetGroup(request) .whenComplete((response, exception) -> { if (exception != null) { Throwable cause = exception.getCause(); if (cause instanceof ServiceQuotaExceededException) { throw new CompletionException("The operation was denied because the request would exceed the maximum quota.", cause); } throw new CompletionException("Failed to create subnet group: " + exception.getMessage(), exception); } }) .thenApply(response -> { String name = response.dbSubnetGroup().dbSubnetGroupName(); String arn = response.dbSubnetGroup().dbSubnetGroupArn(); logger.info("Subnet group created: " + name); return arn; }); } ``` + For API details, see [CreateDBSubnetGroup](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/CreateDBSubnetGroup) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 create_subnet_group(neptune_client, group_name: str): """ Creates a Neptune DB subnet group and returns its name and ARN. Args: neptune_client (boto3.client): The Neptune client object. group_name (str): The desired name of the subnet group. Returns: tuple(str, str): (subnet_group_name, subnet_group_arn) Raises: RuntimeError: For quota errors or other AWS-related failures. """ vpc_id = get_default_vpc_id() subnet_ids = get_subnet_ids(vpc_id) request = { 'DBSubnetGroupName': group_name, 'DBSubnetGroupDescription': 'My Neptune subnet group', 'SubnetIds': subnet_ids, 'Tags': [{'Key': 'Environment', 'Value': 'Dev'}] } try: response = neptune_client.create_db_subnet_group(**request) sg = response.get("DBSubnetGroup", {}) name = sg.get("DBSubnetGroupName") arn = sg.get("DBSubnetGroupArn") if not name or not arn: raise RuntimeError("Response missing subnet group name or ARN.") print(f"Subnet group created: {name}") print(f"ARN: {arn}") return name, arn except ClientError as e: code = e.response["Error"]["Code"] msg = e.response["Error"]["Message"] if code == "ServiceQuotaExceededException": print("Subnet group quota exceeded.") raise RuntimeError("Subnet group quota exceeded.") from e else: print(f"AWS error [{code}]: {msg}") raise RuntimeError(f"AWS error [{code}]: {msg}") from e except Exception as e: print(f"Unexpected error creating subnet group '{group_name}': {e}") raise RuntimeError(f"Unexpected error creating subnet group '{group_name}': {e}") from e ``` + For API details, see [CreateDBSubnetGroup](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/CreateDBSubnetGroup) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `CreateGraph` with an AWS SDK The following code examples show how to use `CreateGraph`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Executes the process of creating a new Neptune graph. * * @param client the Neptune graph client used to interact with the Neptune service * @param graphName the name of the graph to be created * @throws NeptuneGraphException if an error occurs while creating the graph */ public static void executeCreateGraph(NeptuneGraphClient client, String graphName) { try { // Create the graph request CreateGraphRequest request = CreateGraphRequest.builder() .graphName(graphName) .provisionedMemory(16) .build(); // Create the graph CreateGraphResponse response = client.createGraph(request); // Extract the graph name and ARN String createdGraphName = response.name(); String graphArn = response.arn(); String graphEndpoint = response.endpoint(); System.out.println("Graph created successfully!"); System.out.println("Graph Name: " + createdGraphName); System.out.println("Graph ARN: " + graphArn); System.out.println("Graph Endpoint: " +graphEndpoint ); } catch (NeptuneGraphException e) { System.err.println("Failed to create graph: " + e.awsErrorDetails().errorMessage()); } finally { client.close(); } } ``` + For API details, see [CreateGraph](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/CreateGraph) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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). ``` """ Running this example. ---------------------------------------------------------------------------------- VPC Networking Requirement: ---------------------------------------------------------------------------------- Amazon Neptune must be accessed from **within the same VPC** as the Neptune cluster. It does not expose a public endpoint, so this code must be executed from: - An **AWS Lambda function** configured to run inside the same VPC - An **EC2 instance** or **ECS task** running in the same VPC - A connected environment such as a **VPN**, **AWS Direct Connect**, or a **peered VPC** """ GRAPH_NAME = "sample-analytics-graph" def main(): config = Config(retries={"total_max_attempts": 1, "mode": "standard"}, read_timeout=None) client = boto3.client("neptune-graph", config=config) execute_create_graph(client, GRAPH_NAME) def execute_create_graph(client, graph_name): try: print("Creating Neptune graph...") response = client.create_graph( graphName=graph_name, provisionedMemory = 16 ) created_graph_name = response.get("name") graph_arn = response.get("arn") graph_endpoint = response.get("endpoint") print("Graph created successfully!") print(f"Graph Name: {created_graph_name}") print(f"Graph ARN: {graph_arn}") print(f"Graph Endpoint: {graph_endpoint}") except ClientError as e: print(f"Failed to create graph: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"Failed to create graph: {str(e)}") except Exception as e: print(f"Unexpected error: {str(e)}") if __name__ == "__main__": main() ``` + For API details, see [CreateGraph](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/CreateGraph) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `DeleteDBCluster` with an AWS SDK The following code examples show how to use `DeleteDBCluster`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Deletes a DB instance asynchronously. * * @param clusterId the identifier of the cluster to delete * @return a {@link CompletableFuture} that completes when the cluster has been deleted */ public CompletableFuture deleteDBClusterAsync(String clusterId) { DeleteDbClusterRequest request = DeleteDbClusterRequest.builder() .dbClusterIdentifier(clusterId) .skipFinalSnapshot(true) .build(); return getAsyncClient().deleteDBCluster(request) .thenAccept(response -> System.out.println("šŸ—‘ļø Deleting DB Cluster: " + clusterId)); } ``` + For API details, see [DeleteDBCluster](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/DeleteDBCluster) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 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 ``` + For API details, see [DeleteDBCluster](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/DeleteDBCluster) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `DeleteDBInstance` with an AWS SDK The following code examples show how to use `DeleteDBInstance`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Deletes a DB instance asynchronously. * * @param instanceId the identifier of the DB instance to be deleted * @return a {@link CompletableFuture} that completes when the DB instance has been deleted */ public CompletableFuture deleteDBInstanceAsync(String instanceId) { DeleteDbInstanceRequest request = DeleteDbInstanceRequest.builder() .dbInstanceIdentifier(instanceId) .skipFinalSnapshot(true) .build(); return getAsyncClient().deleteDBInstance(request) .thenAccept(response -> System.out.println("šŸ—‘ļø Deleting DB Instance: " + instanceId)); } ``` + For API details, see [DeleteDBInstance](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/DeleteDBInstance) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 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 ``` + For API details, see [DeleteDBInstance](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/DeleteDBInstance) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `DeleteDBSubnetGroup` with an AWS SDK The following code examples show how to use `DeleteDBSubnetGroup`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Deletes a subnet group. * * @param subnetGroupName the identifier of the subnet group to delete * @return a {@link CompletableFuture} that completes when the cluster has been deleted */ public CompletableFuture deleteDBSubnetGroupAsync(String subnetGroupName) { DeleteDbSubnetGroupRequest request = DeleteDbSubnetGroupRequest.builder() .dbSubnetGroupName(subnetGroupName) .build(); return getAsyncClient().deleteDBSubnetGroup(request) .thenAccept(response -> logger.info("šŸ—‘ļø Deleting Subnet Group: " + subnetGroupName)); } ``` + For API details, see [DeleteDBSubnetGroup](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/DeleteDBSubnetGroup) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 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 ``` + For API details, see [DeleteDBSubnetGroup](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/DeleteDBSubnetGroup) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `DescribeDBClusters` with an AWS SDK The following code examples show how to use `DescribeDBClusters`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * 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 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); }); } ``` + For API details, see [DescribeDBClusters](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/DescribeDBClusters) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 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 ``` + For API details, see [DescribeDBClusters](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/DescribeDBClusters) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `DescribeDBInstances` with an AWS SDK The following code examples show how to use `DescribeDBInstances`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Checks the status of a Neptune instance recursively until the desired status is reached or a timeout occurs. * * @param instanceId the ID of the Neptune instance to check * @param desiredStatus the desired status of the Neptune instance * @param startTime the start time of the operation, used to calculate the elapsed time * @param future a {@link CompletableFuture} that will be completed when the desired status is reached */ private void checkStatusRecursive(String instanceId, String desiredStatus, long startTime, CompletableFuture future) { DescribeDbInstancesRequest request = DescribeDbInstancesRequest.builder() .dbInstanceIdentifier(instanceId) .build(); getAsyncClient().describeDBInstances(request) .whenComplete((response, exception) -> { if (exception != null) { Throwable cause = exception.getCause(); future.completeExceptionally( new CompletionException("Error checking Neptune instance status", cause) ); return; } List instances = response.dbInstances(); if (instances.isEmpty()) { future.completeExceptionally(new RuntimeException("Instance not found: " + instanceId)); return; } String currentStatus = instances.get(0).dbInstanceStatus(); long elapsedSeconds = (System.currentTimeMillis() - startTime) / 1000; System.out.printf("\r Elapsed: %-20s Status: %-20s", formatElapsedTime((int) elapsedSeconds), currentStatus); System.out.flush(); if (desiredStatus.equalsIgnoreCase(currentStatus)) { System.out.printf("\r Neptune instance reached desired status '%s' after %s.\n", desiredStatus, formatElapsedTime((int) elapsedSeconds)); future.complete(null); } else { CompletableFuture.delayedExecutor(20, TimeUnit.SECONDS) .execute(() -> checkStatusRecursive(instanceId, desiredStatus, startTime, future)); } }); } ``` + For API details, see [DescribeDBInstances](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/DescribeDBInstances) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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 check_instance_status(neptune_client, instance_id: str, desired_status: str): """ Polls the status of a Neptune DB instance until it reaches desired_status. Uses pagination via describe_db_instances ā€” even for a single instance. Raises: ClientError: If describe_db_instances fails (e.g., instance not found). RuntimeError: If timeout expires before reaching desired status. """ paginator = neptune_client.get_paginator('describe_db_instances') start_time = time.time() while True: try: pages = paginator.paginate(DBInstanceIdentifier=instance_id) instances = [] for page in pages: instances.extend(page.get('DBInstances', [])) except ClientError as err: code = err.response["Error"]["Code"] message = err.response["Error"]["Message"] if code == "DBInstanceNotFound": print(f"Instance '{instance_id}' not found. Please verify the instance ID.") else: print(f"Failed to describe DB instance. {code}: {message}") raise current_status = instances[0].get('DBInstanceStatus') if instances else None elapsed = int(time.time() - start_time) print(f"\rElapsed: {format_elapsed_time(elapsed)} Status: {current_status}", end="", flush=True) if current_status and current_status.lower() == desired_status.lower(): print(f"\nInstance '{instance_id}' reached '{desired_status}' in {format_elapsed_time(elapsed)}.") return if elapsed > TIMEOUT_SECONDS: raise RuntimeError(f"Timeout waiting for '{instance_id}' to reach '{desired_status}'") time.sleep(POLL_INTERVAL_SECONDS) ``` + For API details, see [DescribeDBInstances](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/DescribeDBInstances) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `ExecuteGremlinProfileQuery` with an AWS SDK The following code examples show how to use `ExecuteGremlinProfileQuery`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Executes a Gremlin query against an Amazon Neptune database using the provided {@link NeptunedataClient}. * * @param client the {@link NeptunedataClient} instance to use for executing the Gremlin query */ public static void executeGremlinQuery(NeptunedataClient client) { try { System.out.println("Querying Neptune..."); ExecuteGremlinQueryRequest request = ExecuteGremlinQueryRequest.builder() .gremlinQuery("g.V().has('code', 'ANC')") .build(); ExecuteGremlinQueryResponse response = client.executeGremlinQuery(request); System.out.println("Full Response:"); System.out.println(response); // Retrieve and print the result if (response.result() != null) { System.out.println("Query Result:"); System.out.println(response.result().toString()); } else { System.out.println("No result returned from the query."); } } catch (NeptunedataException e) { System.err.println("Error calling Neptune: " + e.awsErrorDetails().errorMessage()); } catch (Exception e) { System.err.println("Unexpected error: " + e.getMessage()); } finally { client.close(); } } ``` + For API details, see [ExecuteGremlinProfileQuery](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/ExecuteGremlinProfileQuery) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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). ``` # Replace this with your actual Neptune endpoint NEPTUNE_ENDPOINT = "https://[Specify Endpoint]:8182" def main(): """ Entry point of the program. Initializes the Neptune client and executes the Gremlin query. """ config = Config(connect_timeout=10, read_timeout=30, retries={'max_attempts': 3}) neptune_client = boto3.client( "neptunedata", endpoint_url=NEPTUNE_ENDPOINT, config=config ) execute_gremlin_query(neptune_client) def execute_gremlin_query(neptune_client): """ Executes a Gremlin query against an Amazon Neptune database. """ try: print("Querying Neptune...") response = neptune_client.execute_gremlin_explain_query( gremlinQuery="g.V().has('code', 'ANC')" ) print("Full Response:") print(response['output'].read().decode('UTF-8')) except ClientError as e: print(f"Error calling Neptune: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"BotoCore error: {str(e)}") except Exception as e: print(f"Unexpected error: {str(e)}") if __name__ == "__main__": main() ``` + For API details, see [ExecuteGremlinProfileQuery](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/ExecuteGremlinProfileQuery) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `ExecuteGremlinQuery` with an AWS SDK The following code examples show how to use `ExecuteGremlinQuery`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Executes a Gremlin PROFILE query using the provided NeptunedataClient. * * @param client The NeptunedataClient instance to be used for executing the Gremlin PROFILE query. */ private static void executeGremlinProfileQuery(NeptunedataClient client) { System.out.println("Executing Gremlin PROFILE query..."); ExecuteGremlinProfileQueryRequest request = ExecuteGremlinProfileQueryRequest.builder() .gremlinQuery("g.V().has('code', 'ANC')") .build(); ExecuteGremlinProfileQueryResponse response = client.executeGremlinProfileQuery(request); if (response.output() != null) { System.out.println("Query Profile Output:"); System.out.println(response.output()); } else { System.out.println("No output returned from the profile query."); } } ``` + For API details, see [ExecuteGremlinQuery](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/ExecuteGremlinQuery) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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). ``` """ Running this example. ---------------------------------------------------------------------------------- VPC Networking Requirement: ---------------------------------------------------------------------------------- Amazon Neptune must be accessed from **within the same VPC** as the Neptune cluster. It does not expose a public endpoint, so this code must be executed from: - An **AWS Lambda function** configured to run inside the same VPC - An **EC2 instance** or **ECS task** running in the same VPC - A connected environment such as a **VPN**, **AWS Direct Connect**, or a **peered VPC** """ # Replace with your actual Neptune endpoint NEPTUNE_ENDPOINT = "https://[Specify-Your-Endpoint]:8182" def main(): """ Entry point of the program. Initializes the Neptune client and runs both EXPLAIN and PROFILE queries. """ config = Config(connect_timeout=10, read_timeout=30, retries={'max_attempts': 3}) neptune_client = boto3.client( "neptunedata", endpoint_url=NEPTUNE_ENDPOINT, config=config ) try: run_profile_query(neptune_client) except ClientError as e: print(f"Neptune error: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"BotoCore error: {str(e)}") except Exception as e: print(f"Unexpected error: {str(e)}") def run_profile_query(neptune_client): """ Runs a PROFILE query on the Neptune graph database. """ print("Running Gremlin PROFILE query...") try: response = neptune_client.execute_gremlin_profile_query( gremlinQuery="g.V().has('code', 'ANC')" ) print("Profile Query Result:") output = response.get("output") if output: print(output.read().decode('utf-8')) else: print("No explain output returned.") except Exception as e: print(f"Failed to execute PROFILE query: {str(e)}") if __name__ == "__main__": main() ``` + For API details, see [ExecuteGremlinQuery](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/ExecuteGremlinQuery) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `ExecuteOpenCypherExplainQuery` with an AWS SDK The following code examples show how to use `ExecuteOpenCypherExplainQuery`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Executes an OpenCypher EXPLAIN query using the provided Neptune data client. * * @param client The Neptune data client to use for the query execution. */ public static void executeGremlinQuery(NeptunedataClient client) { try { System.out.println("Executing OpenCypher EXPLAIN query..."); ExecuteOpenCypherExplainQueryRequest request = ExecuteOpenCypherExplainQueryRequest.builder() .openCypherQuery("MATCH (n {code: 'ANC'}) RETURN n") .explainMode("debug") .build(); ExecuteOpenCypherExplainQueryResponse response = client.executeOpenCypherExplainQuery(request); if (response.results() != null) { System.out.println("Explain Results:"); System.out.println(response.results().asUtf8String()); } else { System.out.println("No explain results returned."); } } catch (NeptunedataException e) { System.err.println("Neptune error: " + e.awsErrorDetails().errorMessage()); } catch (Exception e) { System.err.println("Unexpected error: " + e.getMessage()); } finally { client.close(); } } ``` + For API details, see [ExecuteOpenCypherExplainQuery](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/ExecuteOpenCypherExplainQuery) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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). ``` # Replace with your actual Neptune endpoint URL NEPTUNE_ENDPOINT = "https://:8182" def main(): """ Entry point: Create Neptune client and execute different OpenCypher queries. """ config = Config(connect_timeout=10, read_timeout=30, retries={'max_attempts': 3}) neptune_client = boto3.client( "neptunedata", endpoint_url=NEPTUNE_ENDPOINT, config=config ) execute_open_cypher_query_without_params(neptune_client) execute_open_cypher_query_with_params(neptune_client) execute_open_cypher_explain_query(neptune_client) def execute_open_cypher_query_without_params(client): """ Executes a simple OpenCypher query without parameters. """ try: print("\nRunning OpenCypher query without parameters...") resp = client.execute_open_cypher_query( openCypherQuery="MATCH (n {code: 'ANC'}) RETURN n" ) print("Results:") print(resp['results']) except Exception as e: print(f"Error in simple OpenCypher query: {str(e)}") def execute_open_cypher_query_with_params(client): """ Executes an OpenCypher query using parameters. """ try: print("\nRunning OpenCypher query with parameters...") parameters = {'code': 'ANC'} resp = client.execute_open_cypher_query( openCypherQuery="MATCH (n {code: $code}) RETURN n", parameters=json.dumps(parameters) ) print("Results:") print(resp['results']) except Exception as e: print(f"Error in parameterized OpenCypher query: {str(e)}") def execute_open_cypher_explain_query(client): """ Runs an OpenCypher EXPLAIN query in debug mode. """ try: print("\nRunning OpenCypher EXPLAIN query (debug mode)...") resp = client.execute_open_cypher_explain_query( openCypherQuery="MATCH (n {code: 'ANC'}) RETURN n", explainMode="details" ) results = resp.get('results') if results is None: print("No explain results returned.") else: try: print("Explain Results:") print(results.read().decode('UTF-8')) except Exception as e: print(f"Error in OpenCypher EXPLAIN query: {str(e)}") except ClientError as e: print(f"Neptune error: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"BotoCore error: {str(e)}") except Exception as e: print(f"Unexpected error: {str(e)}") if __name__ == "__main__": main() ``` + For API details, see [ExecuteOpenCypherExplainQuery](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/ExecuteOpenCypherExplainQuery) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `ExecuteQuery` with an AWS SDK The following code examples show how to use `ExecuteQuery`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Executes a Gremlin profile query on the Neptune Analytics graph. * * @param client the {@link NeptuneGraphClient} instance to use for the query * @param graphId the identifier of the graph to execute the query on * * @throws NeptuneGraphException if an error occurs while executing the query on the Neptune Graph * @throws Exception if an unexpected error occurs */ public static void executeGremlinProfileQuery(NeptuneGraphClient client, String graphId) { try { System.out.println("Running openCypher query on Neptune Analytics..."); ExecuteQueryRequest request = ExecuteQueryRequest.builder() .graphIdentifier(graphId) .queryString("MATCH (n {code: 'ANC'}) RETURN n") .language("OPEN_CYPHER") .build(); ResponseInputStream response = client.executeQuery(request); try (BufferedReader reader = new BufferedReader(new InputStreamReader(response, StandardCharsets.UTF_8))) { String result = reader.lines().collect(Collectors.joining("\n")); System.out.println("Query Result:"); System.out.println(result); } catch (Exception e) { System.err.println("Error reading response: " + e.getMessage()); } } catch (NeptuneGraphException e) { System.err.println("NeptuneGraph error: " + e.awsErrorDetails().errorMessage()); } catch (Exception e) { System.err.println("Unexpected error: " + e.getMessage()); } finally { client.close(); } } ``` + For API details, see [ExecuteQuery](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/ExecuteQuery) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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). ``` """ Running this example. ---------------------------------------------------------------------------------- VPC Networking Requirement: ---------------------------------------------------------------------------------- Amazon Neptune must be accessed from **within the same VPC** as the Neptune cluster. It does not expose a public endpoint, so this code must be executed from: - An **AWS Lambda function** configured to run inside the same VPC - An **EC2 instance** or **ECS task** running in the same VPC - A connected environment such as a **VPN**, **AWS Direct Connect**, or a **peered VPC** """ GRAPH_ID = "" def main(): config = Config(retries={"total_max_attempts": 1, "mode": "standard"}, read_timeout=None) client = boto3.client("neptune-graph", config=config) try: print("\n--- Running OpenCypher query without parameters ---") run_open_cypher_query(client, GRAPH_ID) print("\n--- Running OpenCypher query with parameters ---") run_open_cypher_query_with_params(client, GRAPH_ID) print("\n--- Running OpenCypher explain query ---") run_open_cypher_explain_query(client, GRAPH_ID) except Exception as e: print(f"Unexpected error in main: {e}") def run_open_cypher_query(client, graph_id): """ Run an OpenCypher query without parameters. """ try: resp = client.execute_query( graphIdentifier=graph_id, queryString="MATCH (n {code: 'ANC'}) RETURN n", language='OPEN_CYPHER' ) print(resp['payload'].read().decode('UTF-8')) except client.exceptions.InternalServerException as e: print(f"InternalServerException: {e.response['Error']['Message']}") except ClientError as e: print(f"ClientError: {e.response['Error']['Message']}") except Exception as e: # <--- ADD THIS BLOCK print(f"Unexpected error: {e}") def run_open_cypher_query_with_params(client, graph_id): """ Run an OpenCypher query with parameters. """ try: parameters = {'code': 'ANC'} resp = client.execute_query( graphIdentifier=graph_id, queryString="MATCH (n {code: $code}) RETURN n", language='OPEN_CYPHER', parameters=parameters ) print(resp['payload'].read().decode('UTF-8')) except client.exceptions.InternalServerException as e: print(f"InternalServerException: {e.response['Error']['Message']}") except ClientError as e: print(f"ClientError: {e.response['Error']['Message']}") except Exception as e: # <--- ADD THIS BLOCK print(f"Unexpected error: {e}") def run_open_cypher_explain_query(client, graph_id): """ Run an OpenCypher explain query (explainMode = "debug"). """ try: resp = client.execute_query( graphIdentifier=graph_id, queryString="MATCH (n {code: 'ANC'}) RETURN n", language='OPEN_CYPHER', explainMode='DETAILS' ) print(resp['payload'].read().decode('UTF-8')) except ClientError as e: print(f"Neptune error: {e.response['Error']['Message']}") except BotoCoreError as e: print(f"Unexpected Boto3 error: {str(e)}") except Exception as e: # <-- Add this generic catch print(f"Unexpected error: {str(e)}") if __name__ == "__main__": main() ``` + For API details, see [ExecuteQuery](https://docs.aws.amazon.com/goto/boto3/neptune-2014-10-31/ExecuteQuery) in *AWS SDK for Python (Boto3) API Reference*. ------ # Use `StartDBCluster` with an AWS SDK The following code examples show how to use `StartDBCluster`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Starts an Amazon Neptune DB cluster. * * @param clusterIdentifier the unique identifier of the DB cluster to be stopped */ public CompletableFuture startDBClusterAsync(String clusterIdentifier) { StartDbClusterRequest clusterRequest = StartDbClusterRequest.builder() .dbClusterIdentifier(clusterIdentifier) .build(); return getAsyncClient().startDBCluster(clusterRequest) .whenComplete((response, error) -> { if (error != null) { Throwable cause = error.getCause() != null ? error.getCause() : error; if (cause instanceof ResourceNotFoundException) { throw (ResourceNotFoundException) cause; } throw new RuntimeException("Failed to start DB cluster: " + cause.getMessage(), cause); } else { logger.info("DB Cluster starting: " + clusterIdentifier); } }); } ``` + For API details, see [StartDBCluster](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/StartDBCluster) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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*. ------ # Use `StopDBCluster` with an AWS SDK The following code examples show how to use `StopDBCluster`. Action examples are code excerpts from larger programs and must be run in context. You can see this action in context in the following code example: + [Learn the basics](neptune_example_neptune_Scenario_section.md) ------ #### [ Java ] **SDK for Java 2.x** 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/javav2/example_code/neptune#code-examples). ``` /** * Stops an Amazon Neptune DB cluster. * * @param clusterIdentifier the unique identifier of the DB cluster to be stopped */ public CompletableFuture stopDBClusterAsync(String clusterIdentifier) { StopDbClusterRequest clusterRequest = StopDbClusterRequest.builder() .dbClusterIdentifier(clusterIdentifier) .build(); return getAsyncClient().stopDBCluster(clusterRequest) .whenComplete((response, error) -> { if (error != null) { Throwable cause = error.getCause() != null ? error.getCause() : error; if (cause instanceof ResourceNotFoundException) { throw (ResourceNotFoundException) cause; } throw new RuntimeException("Failed to stop DB cluster: " + cause.getMessage(), cause); } else { logger.info("DB Cluster stopped: " + clusterIdentifier); } }); } ``` + For API details, see [StopDBCluster](https://docs.aws.amazon.com/goto/SdkForJavaV2/neptune-2014-10-31/StopDBCluster) in *AWS SDK for Java 2.x API Reference*. ------ #### [ Python ] **SDK for Python (Boto3)** 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*. ------