Neo4j Python Driver

Explicit transactions allow you to run multiple statements within a single transaction and give you direct control over commit and rollback activity. They are created using neo4j.Session.begin_transaction(), which returns a neo4j.Transaction object.

Use explicit transactions when you need to distribute Cypher execution across multiple functions for the same transaction or need to run multiple queries without the automatic retries provided by managed transactions.

import neo4j

def transfer_to_other_bank(driver, customer_id, other_bank_id, amount):
    with driver.session(
        database="neo4j",
        default_access_mode=neo4j.WRITE_ACCESS
    ) as session:
        tx = session.begin_transaction()
        try:
            if not customer_balance_check(tx, customer_id, amount):
                return
            other_bank_transfer_api(customer_id, other_bank_id, amount)
            try:
                decrease_customer_balance(tx, customer_id, amount)
                tx.commit()
            except Exception as e:
                raise
        finally:
            tx.close()  # rolls back if not yet committed

Database activity in the async driver is coordinated through two primary abstractions:

1. AsyncSession: A logical container for one or more causally-related transactional units of work. Sessions provide top-level containment and automatically provide guarantees of causal consistency in a clustered environment.

   • Note: Session creation is lightweight, but sessions are not thread-safe and are not concurrency-safe. Avoid using asyncio utilities like asyncio.shield or asyncio.wait_for on individual session methods, as this can lead to multiple tasks handling the same session concurrently, resulting in undefined behavior.
   • Connections are drawn from the AsyncDriver connection pool as needed.

2. Transactions (AsyncTransaction, AsyncManagedTransaction): A unit of work that is either committed in its entirety or rolled back on failure.

To ensure safety when using asyncio.shield, shield the entire coroutine containing the session lifecycle rather than individual session calls.

async def thats_better(driver):
    async def inner()
        async with driver.session() as session:
            await session.run("RETURN 1")

    await asyncio.shield(inner())

Cypher queries can return entire graph structures. The driver provides specific classes to model these entities. Unlike property values, graph entities cannot be passed in as parameters; you must pass their identity or properties explicitly.

Node (neo4j.graph.Node)

Represents a graph node.

• Access properties: Use node[key] (raises KeyError if missing) or node.get(key).
• Metadata: Access node.id, node.element_id, and node.labels.
• Iteration: len(node) returns the number of properties; iterating over the node yields its properties.

Relationship (neo4j.graph.Relationship)

Represents a connection between nodes.

• Access properties: Use relationship[key] or relationship.get(key).
• Metadata: Access relationship.type, relationship.id, relationship.element_id, relationship.start_node, and relationship.end_node.

Path (neo4j.graph.Path)

Represents a sequence of nodes and relationships.

• Metadata: Access path.nodes, path.relationships, path.start_node, and path.end_node.
• Iteration: len(path) returns the number of relationships; iterating over the path yields its relationships.

The neo4j.AsyncDriver is the central object for any Neo4j-backed asynchronous application. It manages a connection pool from which neo4j.AsyncSession objects borrow connections.

Key Lifecycle Rules:

• Lifetime: Create one top-level AsyncDriver instance that lives for the entire lifetime of your application.
• Concurrency: Driver objects are safe to use in concurrent coroutines, but they are not thread-safe.
• Immutability: Connection details (like the URI) are immutable. To change them, you must create a new driver instance.
• Closing: Calling await driver.close() immediately shuts down all connections in the pool.
• Connectivity: To verify the driver can communicate with the database without running a query, use await driver.verify_connectivity().

from neo4j import AsyncGraphDatabase

class Application:
    def __init__(self, uri, user, password):
        self.driver = AsyncGraphDatabase.driver(uri, auth=(user, password))

    async def close(self):
        await self.driver.close()

The driver allows you to pass extended Python types (like tuple, bytearray, numpy.ndarray, or pandas.DataFrame) as query parameters. The driver will serialize these into standard Bolt types before sending them to the server.

Note: The driver will never return these extended types in results; they will always be returned as their core Bolt type equivalents (e.g., a tuple passed in will return as a list).

import neo4j

# Testing type conversion
with neo4j.GraphDatabase.driver(URI, auth=AUTH) as driver:
    with driver.session() as session:
        type_in = ("foo", "bar")  # Passing a tuple
        result = session.run("RETURN $x", x=type_in)
        type_out = result.single()[0]
        print(type(type_out))  # <class 'list'>
        print(type_out)        # ['foo', 'bar']