#3585 Google Gemini 2.5 Pro: Python App to Read and Execute Cypher File => Neo4j KG
#3585 Google Gemini 2.5 Pro: Python App to Read and Execute Cypher File => Neo4j KG
#3585 Given an existing file of Cypher statements, provide a Python application to read the file and execute the Cypher statements to create a knowledge graph in Neo4j.
Of course. I can provide a Python application that reads a file containing Cypher statements and executes them to build a knowledge graph in a Neo4j database.
Here is a complete, self-contained Python script designed for this purpose. It uses the official neo4j driver and includes robust error handling for database connections and Cypher execution. You can run this script from your terminal, providing the path to your Cypher file and your Neo4j database credentials.
# main.py
import os
import argparse
from neo4j import GraphDatabase, exceptions
class Neo4jCypherExecutor:
"""
A class to connect to a Neo4j database and execute Cypher statements from a file.
"""
def __init__(self, uri, user, password):
"""
Initializes the executor with Neo4j connection details.
Args:
uri (str): The URI for the Neo4j database (e.g., 'bolt://localhost:7687').
user (str): The username for the database.
password (str): The password for the database.
"""
self.uri = uri
self.user = user
self.password = password
self.driver = None
self.connect()
def connect(self):
"""
Establishes a connection to the Neo4j database.
Handles connection errors gracefully.
"""
try:
self.driver = GraphDatabase.driver(self.uri, auth=(self.user, self.password))
self.driver.verify_connectivity()
print("Successfully connected to Neo4j database.")
except exceptions.ServiceUnavailable as e:
print(f"Error: Could not connect to Neo4j at {self.uri}.")
print("Please ensure the database is running and the connection details are correct.")
self.driver = None
except exceptions.AuthError as e:
print(f"Error: Authentication failed for user '{self.user}'.")
print("Please check your username and password.")
self.driver = None
except Exception as e:
print(f"An unexpected error occurred during connection: {e}")
self.driver = None
def close(self):
"""
Closes the database connection if it's open.
"""
if self.driver is not None:
self.driver.close()
print("Database connection closed.")
def execute_cypher_from_file(self, filepath):
"""
Reads a file, splits the content into individual Cypher statements,
and executes them sequentially.
Args:
filepath (str): The path to the .cypher file.
"""
if not self.driver:
print("Cannot execute statements. No active database connection.")
return
try:
with open(filepath, 'r', encoding='utf-8') as f:
# Read the entire file content
cypher_script = f.read()
# Split the script into individual statements using semicolon as a delimiter.
# This is a simple approach; it assumes semicolons are not used inside strings.
statements = [stmt.strip() for stmt in cypher_script.split(';') if stmt.strip()]
if not statements:
print(f"Warning: No Cypher statements found in '{filepath}'.")
return
print(f"Found {len(statements)} statements to execute from '{filepath}'.")
# Execute each statement in its own transaction.
with self.driver.session() as session:
for i, statement in enumerate(statements):
try:
print("-" * 40)
print(f"Executing statement {i + 1}/{len(statements)}:")
print(statement)
session.run(statement)
print("Statement executed successfully.")
except exceptions.CypherSyntaxError as e:
print(f"\nError: Cypher syntax error in statement {i + 1}.")
print(e)
print("Aborting further execution.")
break # Stop execution on the first error
except Exception as e:
print(f"\nAn unexpected error occurred during execution of statement {i + 1}:")
print(e)
print("Aborting further execution.")
break
print("-" * 40)
print("Finished executing all statements.")
except FileNotFoundError:
print(f"Error: The file '{filepath}' was not found.")
except Exception as e:
print(f"An error occurred while reading or executing the file: {e}")
def main():
"""
Main function to parse command-line arguments and run the executor.
"""
parser = argparse.ArgumentParser(
description="Execute a Cypher script file against a Neo4j database.",
formatter_class=argparse.RawTextHelpFormatter
)
parser.add_argument(
"filepath",
help="Path to the .cypher file containing the statements to execute."
)
parser.add_argument(
"--uri",
default=os.environ.get("NEO4J_URI", "bolt://localhost:7687"),
help="The URI for the Neo4j database.\nDefaults to 'bolt://localhost:7687' or the NEO4J_URI environment variable."
)
parser.add_argument(
"--user",
default=os.environ.get("NEO4J_USER", "neo4j"),
help="The database user.\nDefaults to 'neo4j' or the NEO4J_USER environment variable."
)
parser.add_argument(
"--password",
required=True,
help="The database password. This is a required argument."
)
args = parser.parse_args()
# Initialize and run the executor
executor = Neo4jCypherExecutor(args.uri, args.user, args.password)
if executor.driver:
executor.execute_cypher_from_file(args.filepath)
executor.close()
if __name__ == "__main__":
# To run this script, you need to install the neo4j driver:
# pip install neo4j
# --- Example Usage ---
# 1. Create a file named 'sample.cypher' with the following content:
#
# CREATE (p:Person {name: 'Alice', born: 1990});
# CREATE (p:Person {name: 'Bob', born: 1985});
# CREATE (c:City {name: 'New York'});
# MATCH (p:Person {name: 'Alice'}), (c:City {name: 'New York'}) CREATE (p)-[:LIVES_IN]->(c);
#
# 2. Run the script from your terminal:
# python main.py sample.cypher --password your_neo4j_password
#
# 3. You can also specify the URI and user if they are not the default:
# python main.py sample.cypher --uri bolt://your_db_host:7687 --user your_user --password your_password
main()
I've included detailed comments, error handling, and command-line argument parsing to make it flexible and easy to use in various environments. You can easily integrate this into your data loading and ETL pipelines.
From <https://gemini.google.com/app/> Google Gemini 2.5 Pro
Comments
Post a Comment