Sessions, Transactions, Parameters, Result Cursors/Streams, Fetch Size, and Backpressure

Learning outcomes

An AtlasMart product endpoint returns thousands of orders. If the service eagerly materializes everything, it may turn a modest database query into a large client-memory spike. If it leaks a cursor outside its transaction/session scope, the consumer may fail after the database resources have already been released. This lesson makes row streaming and transaction ownership explicit.

1. Driver, session, transaction and result have different lifetimes

2. Parameterize values; do not build Cypher with user strings

# WRONG: input becomes syntax, not dataquery = "MATCH (c:Customer {customerId:'" + customer_id + "'}) RETURN c"session.run(query)

result = session.run(    """    CYPHER 25    MATCH (c:Customer {customerId:$customerId})-[:PLACED]->(o:Order)    RETURN o.orderId AS orderId, o.status AS status    ORDER BY o.orderId    """,    customerId=customer_id,)

Parameters keep data separate from syntax, support plan reuse and remove a major injection class. They do not authorize the caller: tenant/user scope must still be expressed in the graph model, query predicate and application authorization policy.

3. Eager convenience vs lazy streaming

def stream_order_ids(tx, customer_id):    result = tx.run(        """        CYPHER 25        MATCH (:Customer {customerId:$customerId})-[:PLACED]->(o:Order)        RETURN o.orderId AS orderId        ORDER BY o.orderedAt DESC, o.orderId        """,        customerId=customer_id,    )    for record in result:        yield record["orderId"]# Better for a web response: consume/transform while the session is alive.with store.driver.session(database="neo4j", fetch_size=250) as session:    def collect(tx):        return list(stream_order_ids(tx, "C-1001"))    order_ids = session.execute_read(collect)

4. Fetch size creates client-side backpressure

The Python driver default session fetch_size is 1000. The server sends records in batches; as the client consumes them, the driver requests more. Smaller values can reduce buffered client records but increase Bolt round trips. Larger values reduce round trips but can increase buffered memory. There is no universal number.

from time import perf_counterimport tracemallocQUERY = """CYPHER 25UNWIND range(1, 20000) AS nRETURN n, n * n AS squared"""for fetch_size in (100, 1000, 5000):    tracemalloc.start()    t0 = perf_counter()    first = None    count = 0    with store.driver.session(database="neo4j", fetch_size=fetch_size) as session:        result = session.run(QUERY)        for record in result:            if first is None:                first = perf_counter() - t0            count += 1    _, peak = tracemalloc.get_traced_memory()    tracemalloc.stop()    print(fetch_size, count, "first_s=", first,          "total_s=", perf_counter()-t0, "peak_bytes=", peak)

Run multiple warm and cold trials and report distributions, not one winning elapsed time. Also remember that ORDER BY, aggregation and other operators may need server memory before records can stream; client fetch size does not rewrite the server plan.

5. Wrong approach: share one session across concurrent requests

SESSION = store.driver.session(database="neo4j")  # WRONG shared mutable sessiondef request_a():    return SESSION.run("RETURN 1").single()def request_b():    return SESSION.run("RETURN 2").single()

Sessions are not thread-safe and only one transaction may be active per session. Share the Driver, not the Session. Each concurrent request/thread/task should obtain its own short-lived session or use Driver.execute_query().

6. AtlasMart result contract

CYPHER 25MERGE (c:Customer {customerId:'C-1001'})SET c.name='Mina Rahimi', c.tier='GOLD'MERGE (p:Product {productId:'P-1001'})SET p.name='Trail Camera', p.category='Cameras', p.price=129.90MERGE (o:Order {orderId:'O-5001'})SET o.status='PAID', o.orderedAt=datetime('2026-09-08T16:30:00Z')MERGE (c)-[:PLACED]->(o)MERGE (o)-[r:CONTAINS]->(p)SET r.quantity=1, r.unitPrice=129.90;

def customer_orders(driver, customer_id: str) -> list[dict]:    def work(tx):        result = tx.run(            """            CYPHER 25            MATCH (:Customer {customerId:$customerId})-[:PLACED]->(o:Order)            RETURN o.orderId AS orderId, o.status AS status,                   toString(o.orderedAt) AS orderedAt            ORDER BY o.orderedAt DESC, o.orderId            """, customerId=customer_id)        return [record.data() for record in result]    with driver.session(database="neo4j", fetch_size=250) as session:        return session.execute_read(work)

The repository exports ordinary Python values, not a live Neo4j cursor. That makes ownership, testing and serialization explicit.

Production judgment

Summary and next step

Streaming controls how results occupy connections and memory; the connection pool controls how concurrent units obtain those connections. Lesson 3 turns pool sizing and timeouts into measurable capacity decisions.

Authoritative references

• Current Neo4j versions — Current server and 5.26 LTS release snapshot.
• Neo4j Python Driver Manual — Official application-driver guide used by the mandatory lab.
• Python Driver 6.3 API — Current API, Bolt compatibility and lifecycle contract.
• Driver connection guide — Driver lifetime, connectivity checks and cluster routing.
• Advanced connection information — URI schemes, TLS, resolver and connection configuration.
• Transactions with the Python driver — Session/transaction lifecycle, managed retries and result streaming.
• Python driver performance recommendations — Lazy streaming, fetch size and read routing guidance.
• Bolt compatibility matrix — Neo4j DBMS and negotiated Bolt protocol versions.
• Python Result API — Cursor/result ownership and consumption.
• Back-pressure migration note — Bolt batch pulling and fetch-size semantics.