# Neo4j Python Driver The Neo4j Python Driver is the official driver for connecting Python applications to Neo4j graph databases using the Bolt protocol. It provides both synchronous and asynchronous APIs for executing Cypher queries, managing transactions, and handling results. The driver supports Python 3.10+ and works with Neo4j server versions through automatic protocol negotiation. The driver offers multiple transaction patterns including auto-commit transactions, explicit transactions, and managed transaction functions with automatic retry logic for transient errors. It supports connection pooling, causal consistency through bookmarks, cluster routing for high availability deployments, and provides type mappings for Neo4j's spatial, temporal, and graph data types. ## Creating a Driver Connection Create a driver instance to establish a connection to a Neo4j database. The driver manages connection pooling and should be reused throughout the application lifecycle. ```python from neo4j import GraphDatabase # Basic connection with authentication URI = "neo4j://localhost:7687" AUTH = ("neo4j", "password") driver = GraphDatabase.driver(URI, auth=AUTH) # Verify connectivity driver.verify_connectivity() # Get server information server_info = driver.get_server_info() print(f"Connected to: {server_info.agent}") print(f"Protocol version: {server_info.protocol_version}") # Always close the driver when done driver.close() # Or use as context manager (recommended) with GraphDatabase.driver(URI, auth=AUTH) as driver: driver.verify_connectivity() # Use driver here... # Driver automatically closed ``` ## Authentication Methods Configure authentication using built-in helper functions for various authentication schemes. ```python from neo4j import GraphDatabase, basic_auth, bearer_auth, kerberos_auth, custom_auth # Basic authentication (username/password) auth = basic_auth("neo4j", "password") driver = GraphDatabase.driver("neo4j://localhost:7687", auth=auth) # Tuple shorthand for basic auth driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Bearer token authentication (SSO) auth = bearer_auth("base64_encoded_token") driver = GraphDatabase.driver("neo4j://localhost:7687", auth=auth) # Kerberos authentication auth = kerberos_auth("base64_encoded_ticket") driver = GraphDatabase.driver("neo4j://localhost:7687", auth=auth) # Custom authentication scheme auth = custom_auth( principal="neo4j", credentials="password", realm="native", scheme="basic", custom_param="value" ) driver = GraphDatabase.driver("neo4j://localhost:7687", auth=auth) driver.close() ``` ## Driver Configuration Options Configure the driver with various options for connection management, timeouts, and encryption. ```python from neo4j import GraphDatabase, TrustSystemCAs, TrustAll, TrustCustomCAs # Full driver configuration driver = GraphDatabase.driver( "neo4j://localhost:7687", auth=("neo4j", "password"), # Connection pool settings max_connection_pool_size=100, max_connection_lifetime=3600, # seconds connection_acquisition_timeout=60, # seconds # Connection settings connection_timeout=30, # seconds # Retry settings max_transaction_retry_time=30, # seconds # Encryption (for bolt:// scheme) encrypted=True, trusted_certificates=TrustSystemCAs(), # or TrustAll(), TrustCustomCAs(["cert.pem"]) # Custom user agent user_agent="MyApp/1.0", ) # Using secure URI schemes (automatic encryption) # bolt+s:// - encrypted with system CAs # bolt+ssc:// - encrypted with self-signed certificates # neo4j+s:// - routing + encrypted with system CAs # neo4j+ssc:// - routing + encrypted with self-signed certificates driver_secure = GraphDatabase.driver( "neo4j+s://localhost:7687", auth=("neo4j", "password") ) driver.close() driver_secure.close() ``` ## execute_query - Simple Query Execution The simplest way to run queries with automatic session and transaction management. ```python from neo4j import GraphDatabase, RoutingControl driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Simple write query records, summary, keys = driver.execute_query( "CREATE (p:Person {name: $name, age: $age}) RETURN p", name="Alice", age=30, database_="neo4j" ) print(f"Created {summary.counters.nodes_created} nodes") # Read query with routing control records, summary, keys = driver.execute_query( "MATCH (p:Person) WHERE p.age > $min_age RETURN p.name AS name, p.age AS age", min_age=25, database_="neo4j", routing_=RoutingControl.READ # Route to read replica ) for record in records: print(f"{record['name']} is {record['age']} years old") # Using result transformer for custom return types from neo4j import Result record = driver.execute_query( "MATCH (p:Person {name: $name}) RETURN p.age AS age", name="Alice", result_transformer_=Result.single, # Returns single Record or None database_="neo4j" ) if record: print(f"Age: {record['age']}") driver.close() ``` ## Session Management Sessions provide the context for executing transactions and managing database state. ```python from neo4j import GraphDatabase, READ_ACCESS, WRITE_ACCESS driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Basic session usage with context manager with driver.session(database="neo4j") as session: result = session.run("MATCH (n) RETURN count(n) AS count") count = result.single()["count"] print(f"Node count: {count}") # Session with configuration with driver.session( database="neo4j", default_access_mode=READ_ACCESS, fetch_size=1000, # Records fetched per batch impersonated_user="other_user" # Run as different user ) as session: result = session.run("MATCH (n:Person) RETURN n.name AS name") names = [record["name"] for record in result] # Getting bookmarks for causal consistency with driver.session(database="neo4j") as session: session.run("CREATE (p:Person {name: 'Bob'})").consume() bookmarks = session.last_bookmarks() # Using bookmarks in subsequent session with driver.session(database="neo4j", bookmarks=bookmarks) as session: # This session will see the changes from the previous session result = session.run("MATCH (p:Person {name: 'Bob'}) RETURN p") record = result.single() driver.close() ``` ## Auto-commit Transactions (session.run) Execute single queries directly through the session. Best for simple operations that don't require transaction control. ```python from neo4j import GraphDatabase, Query driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Simple query execution result = session.run( "CREATE (p:Person {name: $name}) RETURN p.name AS name", name="Charlie" ) record = result.single() print(f"Created: {record['name']}") # Query with timeout using Query object result = session.run( Query("MATCH (n) RETURN count(n) AS count", timeout=5.0) ) print(f"Count: {result.single()['count']}") # Parameterized query with kwargs result = session.run( "MATCH (p:Person) WHERE p.name = $name AND p.age > $age RETURN p", name="Alice", age=25 ) for record in result: print(record["p"]) # Consume result to get summary result = session.run("CREATE (n:Test)") summary = result.consume() print(f"Query completed in {summary.result_available_after} ms") print(f"Nodes created: {summary.counters.nodes_created}") driver.close() ``` ## Managed Transaction Functions (execute_read/execute_write) Use transaction functions for automatic retry on transient errors. The transaction is committed if the function succeeds, rolled back if it raises an exception. ```python from neo4j import GraphDatabase, ManagedTransaction, unit_of_work driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Define transaction functions def create_person(tx: ManagedTransaction, name: str, age: int) -> str: result = tx.run( "CREATE (p:Person {name: $name, age: $age}) RETURN p.name AS name", name=name, age=age ) return result.single()["name"] def get_people_by_age(tx: ManagedTransaction, min_age: int) -> list[dict]: result = tx.run( "MATCH (p:Person) WHERE p.age >= $min_age RETURN p.name AS name, p.age AS age", min_age=min_age ) return [{"name": record["name"], "age": record["age"]} for record in result] # Execute write transaction with driver.session(database="neo4j") as session: name = session.execute_write(create_person, "David", 35) print(f"Created: {name}") # Execute read transaction with driver.session(database="neo4j") as session: people = session.execute_read(get_people_by_age, 30) for person in people: print(f"{person['name']}: {person['age']}") # Transaction function with timeout and metadata using decorator @unit_of_work(timeout=10.0, metadata={"app": "my_app"}) def create_with_timeout(tx: ManagedTransaction, name: str): result = tx.run("CREATE (p:Person {name: $name}) RETURN p", name=name) return result.single()["p"] with driver.session(database="neo4j") as session: person = session.execute_write(create_with_timeout, "Eve") driver.close() ``` ## Explicit Transactions (begin_transaction) Full control over transaction lifecycle for complex operations requiring manual commit/rollback. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Using transaction as context manager (auto-commit on success, rollback on exception) with driver.session(database="neo4j") as session: with session.begin_transaction() as tx: tx.run("CREATE (a:Person {name: 'Frank'})") tx.run("CREATE (b:Person {name: 'Grace'})") tx.run( "MATCH (a:Person {name: 'Frank'}), (b:Person {name: 'Grace'}) " "CREATE (a)-[:KNOWS]->(b)" ) # Transaction commits automatically when exiting with block # Manual transaction control with driver.session(database="neo4j") as session: tx = session.begin_transaction( metadata={"operation": "batch_import"}, timeout=30.0 ) try: tx.run("CREATE (n:Node {id: 1})") tx.run("CREATE (n:Node {id: 2})") # Check some condition result = tx.run("MATCH (n:Node) RETURN count(n) AS count") count = result.single()["count"] if count >= 2: tx.commit() print("Transaction committed") else: tx.rollback() print("Transaction rolled back") except Exception as e: tx.rollback() raise driver.close() ``` ## Working with Results Process query results with various methods for different use cases. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Setup test data session.run("MATCH (n) DETACH DELETE n").consume() session.run( "UNWIND range(1, 5) AS i CREATE (p:Person {name: 'Person' + toString(i), age: 20 + i})" ).consume() # Iterate over results result = session.run("MATCH (p:Person) RETURN p.name AS name, p.age AS age") for record in result: print(f"{record['name']}: {record['age']}") # Get single record (warns if multiple, None if empty) result = session.run("MATCH (p:Person {name: 'Person1'}) RETURN p.name AS name") record = result.single() if record: print(f"Found: {record['name']}") # Get single record (strict - raises if not exactly one) result = session.run("MATCH (p:Person {name: 'Person1'}) RETURN p.name AS name") record = result.single(strict=True) print(f"Found: {record['name']}") # Fetch specific number of records result = session.run("MATCH (p:Person) RETURN p.name AS name ORDER BY p.name") first_two = result.fetch(2) print(f"First two: {[r['name'] for r in first_two]}") # Peek at next record without consuming result = session.run("MATCH (p:Person) RETURN p.name AS name ORDER BY p.name") next_record = result.peek() print(f"Next will be: {next_record['name']}") # Get values as list result = session.run("MATCH (p:Person) RETURN p.name AS name ORDER BY p.name") names = result.value("name") # List of name values print(f"Names: {names}") # Get all values as list of lists result = session.run("MATCH (p:Person) RETURN p.name, p.age ORDER BY p.name") all_values = result.values() # [[name1, age1], [name2, age2], ...] # Convert to list of dictionaries result = session.run("MATCH (p:Person) RETURN p.name AS name, p.age AS age") data = result.data() print(f"Data: {data}") # Get result keys result = session.run("MATCH (p:Person) RETURN p.name AS name, p.age AS age") print(f"Keys: {result.keys()}") # Convert to EagerResult (buffers all records) result = session.run("MATCH (p:Person) RETURN p.name AS name") eager = result.to_eager_result() print(f"Records: {eager.records}, Keys: {eager.keys}") driver.close() ``` ## Result Summary and Counters Access query execution statistics and metadata through the result summary. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: result = session.run( "CREATE (a:Person {name: 'Test1'})-[r:KNOWS]->(b:Person {name: 'Test2'}) " "SET a.updated = true " "RETURN a, r, b" ) summary = result.consume() # Query timing print(f"Query available after: {summary.result_available_after} ms") print(f"Query consumed after: {summary.result_consumed_after} ms") # Counters for write operations counters = summary.counters print(f"Nodes created: {counters.nodes_created}") print(f"Nodes deleted: {counters.nodes_deleted}") print(f"Relationships created: {counters.relationships_created}") print(f"Relationships deleted: {counters.relationships_deleted}") print(f"Properties set: {counters.properties_set}") print(f"Labels added: {counters.labels_added}") print(f"Labels removed: {counters.labels_removed}") print(f"Indexes added: {counters.indexes_added}") print(f"Indexes removed: {counters.indexes_removed}") print(f"Constraints added: {counters.constraints_added}") print(f"Constraints removed: {counters.constraints_removed}") print(f"Contains updates: {counters.contains_updates}") print(f"Contains system updates: {counters.contains_system_updates}") # Server info print(f"Server: {summary.server.agent}") print(f"Database: {summary.database}") print(f"Query type: {summary.query_type}") # Query profile (if PROFILE was used) result = session.run("PROFILE MATCH (n) RETURN count(n)") summary = result.consume() if summary.profile: print(f"Profile: {summary.profile}") if summary.plan: print(f"Plan: {summary.plan}") driver.close() ``` ## Working with Graph Types (Node, Relationship, Path) Handle Neo4j graph entities returned from queries. ```python from neo4j import GraphDatabase from neo4j.graph import Node, Relationship, Path driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Setup session.run("MATCH (n) DETACH DELETE n").consume() session.run( "CREATE (a:Person:Employee {name: 'Alice', age: 30})" "-[:KNOWS {since: 2020}]->" "(b:Person {name: 'Bob', age: 25})" "-[:KNOWS {since: 2021}]->" "(c:Person {name: 'Charlie', age: 35})" ).consume() # Working with Nodes result = session.run("MATCH (p:Person {name: 'Alice'}) RETURN p") record = result.single() node: Node = record["p"] print(f"Element ID: {node.element_id}") print(f"Labels: {node.labels}") # frozenset({'Person', 'Employee'}) print(f"Properties: {dict(node)}") print(f"Name: {node['name']}") print(f"Age: {node.get('age', 0)}") # Working with Relationships result = session.run( "MATCH (a:Person {name: 'Alice'})-[r:KNOWS]->(b:Person) RETURN r" ) record = result.single() rel: Relationship = record["r"] print(f"Relationship type: {rel.type}") print(f"Start node: {rel.start_node}") print(f"End node: {rel.end_node}") print(f"Properties: {dict(rel)}") print(f"Since: {rel['since']}") # Working with Paths result = session.run( "MATCH path = (a:Person {name: 'Alice'})-[:KNOWS*]->(c:Person {name: 'Charlie'}) " "RETURN path" ) record = result.single() path: Path = record["path"] print(f"Path length: {len(path)}") print(f"Start node: {path.start_node['name']}") print(f"End node: {path.end_node['name']}") print(f"Nodes in path: {[n['name'] for n in path.nodes]}") print(f"Relationships: {[r.type for r in path.relationships]}") # Access the graph from results result = session.run( "MATCH (a:Person)-[r:KNOWS]->(b:Person) RETURN a, r, b" ) graph = result.graph() print(f"Nodes in graph: {len(graph.nodes)}") print(f"Relationships in graph: {len(graph.relationships)}") for node in graph.nodes: print(f" Node: {node['name']}") driver.close() ``` ## Working with Records Access record data using various methods. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: result = session.run( "RETURN 'Alice' AS name, 30 AS age, ['Python', 'Neo4j'] AS skills" ) record = result.single() # Access by key name print(f"Name: {record['name']}") print(f"Age: {record['age']}") # Access by index print(f"First value: {record[0]}") print(f"Second value: {record[1]}") # Get value with default print(f"Email: {record.get('email', 'not provided')}") # Get single value print(f"Value of 'name': {record.value('name')}") print(f"Value at index 0: {record.value(0)}") print(f"Value with default: {record.value('missing', 'default')}") # Get multiple values print(f"All values: {record.values()}") print(f"Specific values: {record.values('name', 'age')}") # Get keys print(f"Keys: {record.keys()}") # Get items as key-value pairs print(f"Items: {list(record.items())}") # Convert to dictionary print(f"As dict: {record.data()}") print(f"Specific fields: {record.data('name', 'age')}") # Check if key exists print(f"Has 'name': {'name' in record}") # Iterate over record for key in record.keys(): print(f"{key}: {record[key]}") driver.close() ``` ## Pandas DataFrame Integration Convert results directly to pandas DataFrames for data analysis. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Setup test data session.run("MATCH (n) DETACH DELETE n").consume() session.run(""" CREATE (a:Person {name: 'Alice', age: 30, joined: date('2020-01-15')}) CREATE (b:Person {name: 'Bob', age: 25, joined: date('2021-06-20')}) CREATE (c:Person {name: 'Charlie', age: 35, joined: date('2019-03-10')}) CREATE (a)-[:KNOWS {since: 2020}]->(b) CREATE (b)-[:KNOWS {since: 2021}]->(c) """).consume() # Basic DataFrame conversion result = session.run("MATCH (p:Person) RETURN p.name AS name, p.age AS age") df = result.to_df() print(df) # name age # 0 Alice 30 # 1 Bob 25 # 2 Charlie 35 # DataFrame with expanded graph types result = session.run("MATCH (p:Person) RETURN p") df = result.to_df(expand=True) print(df.columns.tolist()) # ['p().prop.name', 'p().prop.age', 'p().prop.joined', 'p().element_id', 'p().labels'] # Parse date/datetime columns result = session.run("MATCH (p:Person) RETURN p.name AS name, p.joined AS joined") df = result.to_df(parse_dates=True) print(df.dtypes) # joined column will be pandas Timestamp driver.close() ``` ## Bookmark Manager for Causal Consistency Manage bookmarks across sessions to ensure causal consistency in a cluster. ```python from neo4j import GraphDatabase driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) # Create a bookmark manager bookmark_manager = GraphDatabase.bookmark_manager() # All sessions using this manager will see each other's writes with driver.session(database="neo4j", bookmark_manager=bookmark_manager) as session: session.run("CREATE (p:Person {name: 'Alice'})").consume() # This session sees the write from above with driver.session(database="neo4j", bookmark_manager=bookmark_manager) as session: result = session.run("MATCH (p:Person {name: 'Alice'}) RETURN p") record = result.single() # Guaranteed to find Alice # execute_query uses its own bookmark manager by default driver.execute_query("CREATE (p:Person {name: 'Bob'})", database_="neo4j") # Chain execute_query with session using the driver's bookmark manager with driver.session( database="neo4j", bookmark_manager=driver.execute_query_bookmark_manager ) as session: # This session sees Bob created by execute_query result = session.run("MATCH (p:Person {name: 'Bob'}) RETURN p") record = result.single() driver.close() ``` ## Async Driver Usage Use the async driver for asyncio-based applications. ```python import asyncio from neo4j import AsyncGraphDatabase, RoutingControl async def main(): driver = AsyncGraphDatabase.driver( "neo4j://localhost:7687", auth=("neo4j", "password") ) # Verify connectivity await driver.verify_connectivity() # Simple query with execute_query records, summary, keys = await driver.execute_query( "CREATE (p:Person {name: $name}) RETURN p.name AS name", name="AsyncAlice", database_="neo4j" ) print(f"Created: {records[0]['name']}") # Session-based operations async with driver.session(database="neo4j") as session: # Auto-commit query result = await session.run( "MATCH (p:Person) RETURN p.name AS name" ) async for record in result: print(f"Found: {record['name']}") # Transaction function async def create_person(tx, name: str): result = await tx.run( "CREATE (p:Person {name: $name}) RETURN p.name AS name", name=name ) record = await result.single() return record["name"] name = await session.execute_write(create_person, "AsyncBob") print(f"Created in tx: {name}") # Explicit transaction async with driver.session(database="neo4j") as session: async with await session.begin_transaction() as tx: await tx.run("CREATE (p:Person {name: 'AsyncCharlie'})") # Commits automatically await driver.close() # Run the async code asyncio.run(main()) ``` ## Error Handling Handle various driver and database errors appropriately. ```python from neo4j import GraphDatabase from neo4j.exceptions import ( Neo4jError, ClientError, CypherSyntaxError, ConstraintError, AuthError, ServiceUnavailable, SessionExpired, TransientError, DriverError, ConfigurationError, ResultConsumedError, ) driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) try: with driver.session(database="neo4j") as session: # Syntax error try: session.run("INVALID CYPHER SYNTAX").consume() except CypherSyntaxError as e: print(f"Syntax error: {e.message}") print(f"Error code: {e.code}") # Constraint violation try: session.run("CREATE CONSTRAINT FOR (p:Person) REQUIRE p.id IS UNIQUE").consume() session.run("CREATE (p:Person {id: 1})").consume() session.run("CREATE (p:Person {id: 1})").consume() # Duplicate except ConstraintError as e: print(f"Constraint violation: {e.message}") # Accessing consumed result result = session.run("RETURN 1") result.consume() try: list(result) # Result already consumed except ResultConsumedError as e: print(f"Result consumed: {e}") except AuthError as e: print(f"Authentication failed: {e.message}") except ServiceUnavailable as e: print(f"Database unavailable: {e}") print(f"Retryable: {e.is_retryable()}") except TransientError as e: print(f"Transient error (retryable): {e.message}") print(f"Retryable: {e.is_retryable()}") except Neo4jError as e: print(f"Database error: {e.message}") print(f"Code: {e.code}") print(f"GQL Status: {e.gql_status}") except DriverError as e: print(f"Driver error: {e}") finally: driver.close() ``` ## Temporal Types Work with Neo4j's temporal data types. ```python from neo4j import GraphDatabase from neo4j.time import Date, Time, DateTime, Duration import pytz driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Create Date date = Date(2024, 1, 15) session.run("CREATE (e:Event {date: $date})", date=date).consume() # Create Time time = Time(14, 30, 0, nanosecond=500000000) # 14:30:00.5 session.run("CREATE (e:Event {time: $time})", time=time).consume() # Create DateTime with timezone tz = pytz.timezone('US/Eastern') dt = DateTime(2024, 1, 15, 14, 30, 0, tzinfo=tz) session.run("CREATE (e:Event {datetime: $datetime})", datetime=dt).consume() # Create Duration duration = Duration(months=1, days=5, seconds=3600) # 1 month, 5 days, 1 hour session.run("CREATE (e:Event {duration: $duration})", duration=duration).consume() # Query temporal values result = session.run(""" MATCH (e:Event) RETURN e.date AS date, e.time AS time, e.datetime AS datetime, e.duration AS duration """) for record in result: if record["date"]: d: Date = record["date"] print(f"Date: {d.year}-{d.month}-{d.day}") if record["time"]: t: Time = record["time"] print(f"Time: {t.hour}:{t.minute}:{t.second}") if record["datetime"]: dt: DateTime = record["datetime"] print(f"DateTime: {dt.iso_format()}") if record["duration"]: dur: Duration = record["duration"] print(f"Duration: {dur.months} months, {dur.days} days") driver.close() ``` ## Spatial Types Work with Neo4j's spatial data types (Point). ```python from neo4j import GraphDatabase from neo4j.spatial import Point, CartesianPoint, WGS84Point driver = GraphDatabase.driver("neo4j://localhost:7687", auth=("neo4j", "password")) with driver.session(database="neo4j") as session: # Create 2D Cartesian point cart_point = CartesianPoint((1.5, 2.5)) session.run("CREATE (l:Location {pos: $pos})", pos=cart_point).consume() # Create 3D Cartesian point cart_3d = CartesianPoint((1.0, 2.0, 3.0)) session.run("CREATE (l:Location {pos3d: $pos})", pos=cart_3d).consume() # Create WGS84 point (latitude/longitude) geo_point = WGS84Point((-122.4194, 37.7749)) # San Francisco (lon, lat) session.run("CREATE (l:Location {geo: $geo})", geo=geo_point).consume() # Create WGS84 3D point (with altitude) geo_3d = WGS84Point((-122.4194, 37.7749, 100.0)) # With altitude session.run("CREATE (l:Location {geo3d: $geo})", geo=geo_3d).consume() # Query spatial values result = session.run("MATCH (l:Location) RETURN l.pos AS pos, l.geo AS geo") for record in result: if record["pos"]: p: Point = record["pos"] print(f"Cartesian: x={p.x}, y={p.y}, srid={p.srid}") if record["geo"]: g: Point = record["geo"] print(f"WGS84: lon={g.x}, lat={g.y}, srid={g.srid}") # Distance calculation using Cypher result = session.run(""" WITH point({x: 0, y: 0}) AS p1, point({x: 3, y: 4}) AS p2 RETURN point.distance(p1, p2) AS distance """) print(f"Distance: {result.single()['distance']}") driver.close() ``` ## Summary The Neo4j Python Driver provides a comprehensive API for interacting with Neo4j databases from Python applications. The key usage patterns include: `driver.execute_query()` for simple, one-off queries with automatic session and transaction management; `session.run()` for auto-commit transactions; `session.execute_read()` and `session.execute_write()` for managed transaction functions with automatic retry; and `session.begin_transaction()` for explicit transaction control. The driver supports both synchronous and asynchronous programming models through separate `GraphDatabase` and `AsyncGraphDatabase` entry points. For production applications, the recommended patterns are: use `execute_query()` for simple operations that don't require fine-grained control; use transaction functions (`execute_read`/`execute_write`) for operations that benefit from automatic retry on transient failures; use explicit transactions only when you need manual control over commit/rollback timing; always close drivers, sessions, and results using context managers (`with` statements); configure connection pools appropriately for your workload; and use bookmark managers when causal consistency across sessions is required.