Build a Repository/Service Layer that Keeps Cypher Parameterized, Transactional, Testable, and Measurable

from __future__ import annotationsfrom dataclasses import dataclassfrom neo4j import Driver, Query, unit_of_work@dataclass(frozen=True)class OrderView:    order_id: str    status: strclass AtlasMartRepository:    def __init__(self, driver: Driver, database: str = "neo4j"):        self._driver = driver        self._database = database    def get_order(self, order_id: str, request_id: str) -> OrderView | None:        query = Query(            """            CYPHER 25            MATCH (o:Order {orderId:$orderId})            RETURN o.orderId AS orderId, o.status AS status            """,            timeout=3.0,            metadata={"app":"atlasmart-api", "request_id":request_id,                      "operation":"get-order"},        )        with self._driver.session(database=self._database) as session:            result = session.run(query, orderId=order_id)            record = result.single()            summary = result.consume()            if record is None:                return None            return OrderView(record["orderId"], record["status"])    @staticmethod    @unit_of_work(timeout=5.0)    def _upsert_order_tx(tx, order_id: str, customer_id: str, status: str):        return tx.run(            """            CYPHER 25            MATCH (c:Customer {customerId:$customerId})            MERGE (o:Order {orderId:$orderId})            ON CREATE SET o.createdAt=datetime()            SET o.status=$status            MERGE (c)-[:PLACED]->(o)            RETURN o.orderId AS orderId, o.status AS status            """,            orderId=order_id, customerId=customer_id, status=status,        ).single(strict=True).data()    def upsert_order(self, order_id: str, customer_id: str, status: str):        with self._driver.session(database=self._database) as session:            return session.execute_write(                self._upsert_order_tx, order_id, customer_id, status)

import osfrom neo4j import GraphDatabasefrom atlasmart_repository import AtlasMartRepositoryURI = os.getenv("NEO4J_URI", "bolt://127.0.0.1:7687")AUTH = (os.getenv("NEO4J_USER", "neo4j"),        os.getenv("NEO4J_PASSWORD", "atlasmart-course-2026"))driver = GraphDatabase.driver(    URI, auth=AUTH,    max_connection_pool_size=20,    connection_acquisition_timeout=5.0,    connection_timeout=5.0,    max_connection_lifetime=1800.0,    max_transaction_retry_time=15.0,)try:    driver.verify_connectivity()    repo = AtlasMartRepository(driver, "neo4j")    print(repo.get_order("O-5001", "req-local-001"))    print(repo.upsert_order("O-SVC-1301", "C-1001", "PAID"))finally:    driver.close()

CYPHER 25CREATE CONSTRAINT customer_id IF NOT EXISTSFOR (c:Customer) REQUIRE c.customerId IS UNIQUE;CREATE CONSTRAINT order_id IF NOT EXISTSFOR (o:Order) REQUIRE o.orderId IS UNIQUE;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;

repo = AtlasMartRepository(driver, "neo4j")assert repo.get_order("missing", "test-missing") is Nonefirst = repo.upsert_order("O-SVC-1301", "C-1001", "PAID")second = repo.upsert_order("O-SVC-1301", "C-1001", "PAID")assert first == second == {"orderId":"O-SVC-1301", "status":"PAID"}records, summary, _ = driver.execute_query(    """    CYPHER 25    MATCH (:Customer {customerId:'C-1001'})-[r:PLACED]->(o:Order {orderId:'O-SVC-1301'})    RETURN count(o) AS orders, count(r) AS placedEdges    """, database_="neo4j")row = records[0]assert row["orders"] == 1assert row["placedEdges"] == 1

from time import perf_counterrequest_id = "req-measure-1301"t0 = perf_counter()records, summary, _ = driver.execute_query(    """    CYPHER 25    MATCH (c:Customer {customerId:$id})-[:PLACED]->(o:Order)    RETURN o.orderId AS orderId    ORDER BY o.orderedAt DESC, o.orderId    """,    id="C-1001", database_="neo4j")end_to_end_ms = (perf_counter() - t0) * 1000print({    "request_id": request_id,    "records": len(records),    "result_available_ms": summary.result_available_after,    "result_consumed_ms": summary.result_consumed_after,    "client_end_to_end_ms": end_to_end_ms,})

CYPHER 25MATCH (o:Order)WHERE o.orderId IN ['O-DRIVER-1301','O-BOOKMARK-1','O-SVC-1301']DETACH DELETE o;MATCH (r:Reservation {reservationId:'RSV-1301'}) DETACH DELETE r;