or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authentication.mdconfiguration.mddata-types.mddrivers.mdindex.mdsessions.mdtransactions-results.md
tile.json

tessl/pypi-neo4j

Neo4j Bolt driver for Python providing database connectivity and query execution

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/neo4j@5.28.x

To install, run

npx @tessl/cli install tessl/pypi-neo4j@5.28.0

index.mddocs/

Neo4j Python Driver

The Neo4j Python driver provides connectivity to Neo4j graph databases using the efficient Bolt protocol. It supports both synchronous and asynchronous programming models, offering comprehensive database operations including transactions, sessions, and advanced features like query routing, connection pooling, and causal consistency through bookmarks.

Package Information

  • Package Name: neo4j
  • Package Type: pypi
  • Language: Python
  • Installation: pip install neo4j
  • Requires Python: >=3.7

Core Imports

import neo4j

Most commonly used imports:

from neo4j import GraphDatabase, Driver, Session, Result
from neo4j import basic_auth, Auth
from neo4j import Record, Query, Bookmark

For graph and temporal data types:

from neo4j.graph import Node, Relationship, Path
from neo4j.time import Date, DateTime, Time, Duration
from neo4j.spatial import Point, CartesianPoint, WGS84Point

For exception handling:

from neo4j.exceptions import (
    Neo4jError, ClientError, TransientError, DriverError,
    ServiceUnavailable, AuthError, CypherSyntaxError,
    ResultNotSingleError, SessionExpired
)

For async operations:

from neo4j import AsyncGraphDatabase, AsyncDriver, AsyncSession, AsyncResult
from neo4j import AsyncManagedTransaction, AsyncTransaction

Basic Usage

from neo4j import GraphDatabase, basic_auth

# Create a driver instance
driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=basic_auth("neo4j", "password")
)

# Execute a simple query using execute_query (recommended)
records, summary, keys = driver.execute_query(
    "CREATE (n:Person {name: $name}) RETURN n",
    name="Alice"
)

# Alternative: Use session for more complex operations
with driver.session() as session:
    result = session.run("MATCH (n:Person) RETURN n.name AS name")
    for record in result:
        print(record["name"])

# Always close the driver when done
driver.close()

Async usage:

import asyncio
from neo4j import AsyncGraphDatabase, basic_auth

async def main():
    driver = AsyncGraphDatabase.driver(
        "bolt://localhost:7687", 
        auth=basic_auth("neo4j", "password")
    )
    
    # Execute query asynchronously
    records, summary, keys = await driver.execute_query(
        "MATCH (n:Person) RETURN n.name AS name"
    )
    
    await driver.close()

asyncio.run(main())

Architecture

The Neo4j Python driver follows a hierarchical architecture designed for scalability and flexibility:

  • Driver: Top-level connection manager handling connection pooling, routing, and authentication. Supports both direct connections (BoltDriver) and cluster routing (Neo4jDriver).
  • Session: Logical container for executing work, providing transaction management and causal consistency through bookmarks.
  • Transaction: Unit of work that can be either auto-commit (via run()) or explicit (via begin_transaction()). Supports both read and write operations with automatic retry logic.
  • Result: Iterator-like object for consuming query results, offering both streaming and eager consumption patterns.

This design enables efficient connection reuse, automatic failover in clustered environments, and provides both simple and advanced usage patterns through execute_query shortcuts and explicit transaction management.

Capabilities

Drivers & Connection

Core connection management through GraphDatabase factory methods and Driver classes, supporting both direct server connections and cluster routing with comprehensive configuration options.

class GraphDatabase:
    @staticmethod
    def driver(uri: str, *, auth: Auth = None, **config) -> Driver: ...
    
    @staticmethod
    def bolt_driver(target: str, **config) -> BoltDriver: ...
    
    @staticmethod
    def neo4j_driver(*targets: str, routing_context: dict = None, **config) -> Neo4jDriver: ...

class Driver:
    def session(**config) -> Session: ...
    def execute_query(query: str, parameters: dict = None, **kwargs) -> tuple[list[Record], ResultSummary, list[str]]: ...
    def close() -> None: ...

Drivers & Connection

Sessions

Session management providing contexts for executing work with transaction control, bookmark handling for causal consistency, and support for both read and write operations.

class Session:
    def run(query: str, parameters: dict = None, **kwargs) -> Result: ...
    def execute_read(work: Callable, *args, **kwargs): ...
    def execute_write(work: Callable, *args, **kwargs): ...
    def begin_transaction(**config) -> Transaction: ...
    def close() -> None: ...

Sessions

Transactions & Results

Transaction management and result handling supporting auto-commit transactions, explicit transactions with full ACID properties, and flexible result consumption patterns.

class Transaction:
    def run(query: str, parameters: dict = None, **kwparameters) -> Result: ...
    def commit() -> None: ...
    def rollback() -> None: ...

class Result:
    def consume() -> ResultSummary: ...
    def single(strict: bool = True) -> Record | None: ...
    def data(*args) -> list[dict]: ...
    def to_df() -> pandas.DataFrame: ...

Transactions & Results

Authentication

Authentication system supporting multiple authentication schemes including basic, Kerberos, bearer tokens, and custom authentication mechanisms.

def basic_auth(user: str, password: str, realm: str = None) -> Auth: ...
def kerberos_auth(base64_encoded_ticket: str) -> Auth: ...
def bearer_auth(base64_encoded_token: str) -> Auth: ...
def custom_auth(principal: str, credentials: str, realm: str, scheme: str, **parameters) -> Auth: ...

class Auth:
    scheme: str | None
    principal: str | None
    credentials: str | None
    realm: str | None

Authentication

Data Types

Rich data type system including records, queries with metadata, bookmarks for causal consistency, graph data types (Node, Relationship, Path), temporal types (Date, DateTime, Time, Duration), spatial types (Point variants), and exception hierarchy for error handling.

class Record:
    def keys() -> tuple[str, ...]: ...
    def values() -> tuple: ...
    def get(key: str, default=None): ...
    def data(*keys) -> dict: ...

class Query:
    def __init__(text: str, metadata: dict = None, timeout: float = None): ...
    text: str
    metadata: dict | None
    timeout: float | None

class Node:
    element_id: str
    labels: frozenset[str]
    def get(name: str, default=None): ...

# Neo4j temporal types
class DateTime:
    def __init__(year: int, month: int, day: int, hour: int = 0, ...): ...
    year: int
    month: int
    nanosecond: int

# Spatial types
class Point:
    srid: int | None
    x: float
    y: float

# Exception hierarchy  
class Neo4jError(Exception):
    code: str
    message: str

Data Types

Configuration

Configuration management including SSL/TLS trust settings, connection pooling, timeout configuration, and cluster routing options.

class TrustSystemCAs: ...
class TrustAll: ...
class TrustCustomCAs:
    def __init__(*certs): ...

# Configuration constants
TRUST_SYSTEM_CA_SIGNED_CERTIFICATES: str
TRUST_ALL_CERTIFICATES: str
READ_ACCESS: str
WRITE_ACCESS: str

Configuration