Async Driver APIs, Concurrency Limits, Cancellation, Timeouts, and Resource Cleanup

$env:NEO4J_URI='bolt://127.0.0.1:7687'$env:NEO4J_USER='neo4j'$env:NEO4J_PASSWORD='atlasmart-course-2026'$env:NEO4J_DATABASE='neo4j'py -3 -m venv .venv.\.venv\Scripts\Activate.ps1python -m pip install --upgrade pippython -m pip install neo4j==6.3.0

CYPHER 25CREATE CONSTRAINT customer_id IF NOT EXISTSFOR (c:Customer) REQUIRE c.customerId IS UNIQUE;CREATE CONSTRAINT product_id IF NOT EXISTSFOR (p:Product) REQUIRE p.productId IS UNIQUE;CREATE CONSTRAINT order_id IF NOT EXISTSFOR (o:Order) REQUIRE o.orderId IS UNIQUE;MERGE (c:Customer {customerId:'C-1001'})SET c.name='Mina Rahimi', c.tier='GOLD'MERGE (p1:Product {productId:'P-1001'})SET p1.name='Trail Camera', p1.category='Cameras', p1.price=129.90MERGE (p2:Product {productId:'P-2001'})SET p2.name='Smart Shelf Sensor', p2.category='Store IoT', p2.price=79.50MERGE (o:Order {orderId:'O-5001'})SET o.status='PAID', o.orderedAt=datetime('2026-09-08T16:30:00Z')MERGE (c)-[:PLACED]->(o)MERGE (o)-[r1:CONTAINS]->(p1)SET r1.quantity=1, r1.unitPrice=129.90MERGE (o)-[r2:CONTAINS]->(p2)SET r2.quantity=2, r2.unitPrice=79.50;

from __future__ import annotationsimport osfrom neo4j import AsyncGraphDatabase, RoutingControlclass AsyncAtlasMart:    def __init__(self) -> None:        self.database = os.getenv("NEO4J_DATABASE", "neo4j")        self.driver = AsyncGraphDatabase.driver(            os.getenv("NEO4J_URI", "bolt://127.0.0.1:7687"),            auth=(os.getenv("NEO4J_USER", "neo4j"),                  os.getenv("NEO4J_PASSWORD", "atlasmart-course-2026")),            max_connection_pool_size=20,            connection_acquisition_timeout=5.0,            connection_timeout=5.0,            max_transaction_retry_time=15.0,        )    async def verify(self) -> None:        await self.driver.verify_connectivity()    async def close(self) -> None:        await self.driver.close()    async def get_order(self, order_id: str) -> dict | None:        records, summary, _ = await self.driver.execute_query(            """            CYPHER 25            MATCH (o:Order {orderId:$orderId})            OPTIONAL MATCH (o)-[r:CONTAINS]->(p:Product)            WITH o, collect({productId:p.productId, quantity:r.quantity}) AS lines            RETURN {orderId:o.orderId, status:o.status, lines:lines} AS order            """,            orderId=order_id,            database_=self.database,            routing_=RoutingControl.READ,        )        return records[0]["order"] if records else None

async def wrong(driver, order_ids):    # Every item becomes an immediately scheduled Task.    # A finite pool/server now becomes the queue by accident.    return await asyncio.gather(*[        driver.execute_query(            "CYPHER 25 MATCH (o:Order {orderId:$id}) RETURN o.status",            id=oid, database_="neo4j"        )        for oid in order_ids    ])

import asynciofrom time import perf_counterasync def bounded_read(store, order_ids, concurrency=8):    gate = asyncio.Semaphore(concurrency)    async def one(order_id):        queued_at = perf_counter()        async with gate:            started = perf_counter()            order = await store.get_order(order_id)            finished = perf_counter()        return {            "orderId": order_id,            "queue_ms": (started - queued_at) * 1000,            "service_ms": (finished - started) * 1000,            "found": order is not None,        }    return await asyncio.gather(*(one(x) for x in order_ids))

import asyncioasync def read_lines(driver, order_id):    session = driver.session(database="neo4j", fetch_size=100)    try:        result = await session.run(            """            CYPHER 25            MATCH (:Order {orderId:$id})-[r:CONTAINS]->(p:Product)            RETURN p.productId AS productId, r.quantity AS quantity            ORDER BY productId            """,            id=order_id,        )        return [record.data() async for record in result]    except asyncio.CancelledError:        # Documented purpose: forcefully close the held connection/work.        session.cancel()        raise    finally:        await session.close()

import asynciofrom neo4j import Queryasync def timed_lookup(driver, order_id):    query = Query(        "CYPHER 25 MATCH (o:Order {orderId:$id}) RETURN o.status AS status",        timeout=2.0,        metadata={"app":"atlasmart", "op":"order-status"},    )    async with asyncio.timeout(3.0):        async with driver.session(database="neo4j") as session:            result = await session.run(query, id=order_id)            record = await result.single()            await result.consume()            return None if record is None else record["status"]

SHOW TRANSACTIONSYIELD transactionId, currentQuery, elapsedTime, metaDataWHERE metaData.app = 'atlasmart'RETURN transactionId, elapsedTime, metaData;