Neo4j Python Driver 5.3

The Official Neo4j Driver for Python.

Neo4j versions supported:

  • Neo4j 5.3

  • Neo4j 5.2

  • Neo4j 5.1

  • Neo4j 5.0

  • Neo4j 4.4

Python versions supported:

  • Python 3.11 (added in driver version 5.3.0)

  • Python 3.10

  • Python 3.9

  • Python 3.8

  • Python 3.7

Topics

Installation

To install the latest stable release, use:

python -m pip install neo4j

To install the latest pre-release, use:

python -m pip install --pre neo4j

Note

It is always recommended to install python packages for user space in a virtual environment.

Virtual Environment

To create a virtual environment named sandbox, use:

python -m venv sandbox

To activate the virtual environment named sandbox, use:

source sandbox/bin/activate

To deactivate the current active virtual environment, use:

deactivate

Quick Example

Creating nodes and relationships.

from neo4j import GraphDatabase


URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")


def create_person(tx, name):
    tx.run("CREATE (a:Person {name: $name})", name=name)


def create_friend_of(tx, name, friend):
    tx.run("MATCH (a:Person) WHERE a.name = $name "
           "CREATE (a)-[:KNOWS]->(:Person {name: $friend})",
           name=name, friend=friend)


with GraphDatabase.driver(URI, auth=AUTH) as driver:
    with driver.session() as session:
        session.execute_write(create_person, "Alice")
        session.execute_write(create_friend_of, "Alice", "Bob")
        session.execute_write(create_friend_of, "Alice", "Carl")

Finding nodes.

from neo4j import GraphDatabase


URI = "neo4j://localhost:7687"
AUTH = ("neo4j", "password")


def get_friends_of(tx, name):
    friends = []
    result = tx.run("MATCH (a:Person)-[:KNOWS]->(f) "
                    "WHERE a.name = $name "
                    "RETURN f.name AS friend", name=name)
    for record in result:
        friends.append(record["friend"])
    return friends


with GraphDatabase.driver(URI, auth=AUTH) as driver:
    with driver.session() as session:
        friends = session.execute_read(get_friends_of, "Alice")
        for friend in friends:
            print(friend)

Example Application

import logging
from neo4j import GraphDatabase
from neo4j.exceptions import ServiceUnavailable


class App:

    def __init__(self, uri, user, password):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))

    def close(self):
        # Don't forget to close the driver connection when you are finished with it
        self.driver.close()

    def create_friendship(self, person1_name, person2_name):
        with self.driver.session() as session:
            # Write transactions allow the driver to handle retries and transient errors
            result = session.execute_write(
                self._create_and_return_friendship, person1_name, person2_name)
            for record in result:
                print("Created friendship between: {p1}, {p2}".format(
                    p1=record['p1'], p2=record['p2']))

    @staticmethod
    def _create_and_return_friendship(tx, person1_name, person2_name):

        # To learn more about the Cypher syntax,
        # see https://neo4j.com/docs/cypher-manual/current/

        # The Reference Card is also a good resource for keywords,
        # see https://neo4j.com/docs/cypher-refcard/current/

        query = (
            "CREATE (p1:Person { name: $person1_name }) "
            "CREATE (p2:Person { name: $person2_name }) "
            "CREATE (p1)-[:KNOWS]->(p2) "
            "RETURN p1, p2"
        )
        result = tx.run(query, person1_name=person1_name, person2_name=person2_name)
        try:
            return [{"p1": record["p1"]["name"], "p2": record["p2"]["name"]}
                    for record in result]
        # Capture any errors along with the query and data for traceability
        except ServiceUnavailable as exception:
            logging.error("{query} raised an error: \n {exception}".format(
                query=query, exception=exception))
            raise

    def find_person(self, person_name):
        with self.driver.session() as session:
            result = session.execute_read(self._find_and_return_person, person_name)
            for record in result:
                print("Found person: {record}".format(record=record))

    @staticmethod
    def _find_and_return_person(tx, person_name):
        query = (
            "MATCH (p:Person) "
            "WHERE p.name = $person_name "
            "RETURN p.name AS name"
        )
        result = tx.run(query, person_name=person_name)
        return [record["name"] for record in result]

if __name__ == "__main__":
    # See https://neo4j.com/developer/aura-connect-driver/ for Aura specific connection URL.
    scheme = "neo4j"  # Connecting to Aura, use the "neo4j+s" URI scheme
    host_name = "example.com"
    port = 7687
    url = f"{scheme}://{host_name}:{port}"
    user = "<Username for Neo4j database>"
    password = "<Password for Neo4j database>"
    app = App(url, user, password)
    try:
        app.create_friendship("Alice", "David")
        app.find_person("Alice")
    finally:
        app.close()

Other Information