neo4j-patterns

name	neo4j-patterns
description	Neo4j driver best practices for Python. Use when working with Neo4j connections, Cypher queries, transactions, or GraphRAG implementations. Use when this capability is needed.
metadata	{"author":"arthurfantaci"}

Neo4j Driver Best Practices

Connection URI Schemes

Scheme	Use Case	TLS	Routing
neo4j+s://	Production (Aura, clusters)	✅	✅
neo4j://	Local development	❌	✅
bolt://	Single instance only	❌	❌

ALWAYS use neo4j+s:// for production deployments.

Driver Instance Pattern

✅ GOOD - Create once, reuse:

from contextlib import asynccontextmanager
from neo4j import GraphDatabase

@asynccontextmanager
async def lifespan(app):
    driver = GraphDatabase.driver(
        config.neo4j_uri,
        auth=(config.neo4j_username, config.neo4j_password),
        max_connection_pool_size=5,  # Small for serverless
    )
    driver.verify_connectivity()
    try:
        yield {"driver": driver}
    finally:
        driver.close()

❌ BAD - Creates new driver per request:

def query(cypher):
    driver = GraphDatabase.driver(...)  # DON'T DO THIS
    result = driver.execute_query(cypher)
    driver.close()
    return result

Transaction Functions

✅ GOOD - Use execute_read/execute_write:

def get_articles(driver: Driver, topic: str) -> list[dict]:
    def _query(tx, topic):
        result = tx.run(
            "MATCH (a:Article) WHERE a.topic = $topic RETURN a",
            topic=topic
        )
        return [record.data() for record in result]  # Process IN transaction

    with driver.session(database="neo4j") as session:
        return session.execute_read(_query, topic)

❌ BAD - Auto-commit with session.run:

def get_articles(driver, topic):
    with driver.session() as session:
        result = session.run(f"MATCH (a:Article) WHERE a.topic = '{topic}' RETURN a")
    return [r.data() for r in result]  # Result consumed OUTSIDE transaction!

Query Parameters

✅ GOOD - Always use parameters:

tx.run("MATCH (e) WHERE e.name = $name RETURN e", name=user_input)

❌ BAD - String concatenation (SQL injection risk):

tx.run(f"MATCH (e) WHERE e.name = '{user_input}' RETURN e")

Serverless Optimization

For Vercel/Lambda deployments:

max_connection_pool_size=5,           # Small pool
connection_acquisition_timeout=30.0,   # Cold start tolerance

Result Processing

ALWAYS process results within transaction scope:

def _query(tx, params):
    result = tx.run(query, params)
    return [record.data() for record in result]  # ✅ Inside transaction

# NOT this:
result = session.run(query)
return [record.data() for record in result]  # ❌ Outside transaction

---

Converted and distributed by TomeVault — claim your Tome and manage your conversions.