AWS code-library documentation change
Summary
Added Python code example for ExecuteQuery with VPC access requirements documentation and parameterized query example
Security assessment
Includes VPC networking requirements documentation similar to the Gremlin example. The parameterized query example could help prevent injection attacks, but there's no explicit security advisory about this in the change.
Diff
diff --git a/code-library/latest/ug/neptune_example_neptune_ExecuteQuery_section.md b/code-library/latest/ug/neptune_example_neptune_ExecuteQuery_section.md index 2ae44bc48..0bfa1ad9f 100644 --- a//code-library/latest/ug/neptune_example_neptune_ExecuteQuery_section.md +++ b//code-library/latest/ug/neptune_example_neptune_ExecuteQuery_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 `ExecuteQuery`. +The following code examples show how to use `ExecuteQuery`. @@ -66,0 +67,114 @@ 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). + + + """ + 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 = "<your-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_. + + + +