tessl/pypi-neo4j
Neo4j Bolt driver for Python providing database connectivity and query execution
—
Pending
data-types.mddocs/
Data Types
Rich data type system including records, queries with metadata, bookmarks for causal consistency, and comprehensive support for Neo4j's type system. The data types provide flexible access to query results and enable advanced features like causal consistency and transaction metadata.
Capabilities
Record Class
Immutable key-value collection representing a single result record from a Cypher query, providing dictionary-like access to column values.
class Record:
def keys(self) -> tuple[str, ...]:
"""
Get field names from the record.
Returns:
Tuple of field names as strings
"""
def values(self) -> tuple:
"""
Get field values from the record.
Returns:
Tuple containing all field values in order
"""
def items(self) -> list[tuple[str, Any]]:
"""
Get key-value pairs from the record.
Returns:
List of (key, value) tuples
"""
def get(self, key: str, default=None):
"""
Get value by key with optional default.
Parameters:
- key: Field name to retrieve
- default: Value to return if key not found
Returns:
Field value or default if key doesn't exist
"""
def data(self, *keys: str) -> dict:
"""
Return record as dictionary, optionally filtering by keys.
Parameters:
- *keys: Optional field names to include (all if not specified)
Returns:
Dictionary representation of record
"""
def to_dict(self) -> dict:
"""
Convert record to dictionary.
Returns:
Dictionary containing all key-value pairs
"""
def __getitem__(self, key: str | int) -> Any:
"""
Access field by name or index.
Parameters:
- key: Field name (string) or index (integer)
Returns:
Field value
"""
def __getattr__(self, name: str) -> Any:
"""
Access field as attribute (dot notation).
Parameters:
- name: Field name
Returns:
Field value
"""
def __len__(self) -> int:
"""Get number of fields in the record."""
def __iter__(self) -> Iterator[str]:
"""Iterator over field names."""Example record usage:
from neo4j import GraphDatabase, basic_auth
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
with driver.session() as session:
result = session.run("""
MATCH (n:Person)
RETURN n.name AS name, n.age AS age, n
LIMIT 1
""")
record = result.single()
# Dictionary-style access
print(record["name"])
print(record["age"])
# Attribute-style access
print(record.name)
print(record.age)
# Get with default
email = record.get("email", "no-email@example.com")
# Access by index
first_value = record[0] # First column value
# Iterate over keys and values
for key in record:
print(f"{key}: {record[key]}")
# Convert to dictionary
person_dict = record.data()
print(person_dict) # {'name': 'Alice', 'age': 30, 'n': Node(...)}
# Partial conversion
basic_info = record.data("name", "age")
print(basic_info) # {'name': 'Alice', 'age': 30}Query Class
Query container with attached metadata and configuration for advanced transaction control.
class Query:
def __init__(
self,
text: str,
metadata: dict | None = None,
timeout: float | None = None
):
"""
Create a query with metadata and timeout configuration.
Parameters:
- text: Cypher query string
- metadata: Transaction metadata dictionary
- timeout: Transaction timeout in seconds
"""
@property
def text(self) -> str:
"""
The Cypher query text.
Returns:
Query string
"""
@property
def metadata(self) -> dict | None:
"""
Transaction metadata attached to the query.
Returns:
Metadata dictionary or None
"""
@property
def timeout(self) -> float | None:
"""
Transaction timeout in seconds.
Returns:
Timeout value or None for default
"""Example query usage:
from neo4j import Query, GraphDatabase, basic_auth
# Query with metadata for monitoring
query = Query(
"CREATE (n:Person {name: $name}) RETURN n",
metadata={"application": "user_service", "version": "1.2.3"},
timeout=30.0
)
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
with driver.session() as session:
result = session.run(query, name="Alice")
record = result.single()
print(record["n"]["name"])
# Using @unit_of_work decorator with Query
from neo4j import unit_of_work
@unit_of_work(metadata={"operation": "batch_create"}, timeout=60.0)
def create_users(tx, users):
for user in users:
tx.run("CREATE (n:Person {name: $name})", name=user)
with driver.session() as session:
session.execute_write(create_users, ["Alice", "Bob", "Charlie"])Bookmark Classes
Bookmarks enable causal consistency, ensuring read operations can see the effects of previous write operations across different sessions.
class Bookmark:
"""
Represents a bookmark for causal consistency.
Bookmarks are opaque values returned by transactions to track causality.
"""
class Bookmarks:
"""
Collection of bookmarks for causal consistency across multiple transactions.
"""Bookmark usage for causal consistency:
from neo4j import GraphDatabase, basic_auth
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
# Write operation in one session
with driver.session() as write_session:
write_session.run("CREATE (n:Person {name: 'Alice'})")
# Capture bookmarks after write
bookmarks = write_session.last_bookmarks
# Read operation in another session using bookmarks
with driver.session(bookmarks=bookmarks) as read_session:
result = read_session.run("MATCH (n:Person {name: 'Alice'}) RETURN n")
record = result.single()
# This read is guaranteed to see the Alice node created above
print(f"Found: {record['n']['name']}")
# Bookmark manager for automatic bookmark handling
bookmark_manager = GraphDatabase.bookmark_manager()
# Sessions with bookmark manager automatically handle causal consistency
with driver.session(bookmark_manager=bookmark_manager) as session1:
session1.run("CREATE (n:Person {name: 'Bob'})")
with driver.session(bookmark_manager=bookmark_manager) as session2:
# This session automatically sees changes from session1
result = session2.run("MATCH (n:Person) RETURN count(n) AS count")
count = result.single()["count"]
print(f"Total persons: {count}")BookmarkManager Classes
Bookmark managers automate causal consistency management across sessions and transactions.
class BookmarkManager:
"""
Abstract base class for managing bookmarks automatically.
Handles bookmark storage, retrieval, and causal consistency.
"""
class AsyncBookmarkManager:
"""
Abstract base class for async bookmark management.
Provides the same functionality as BookmarkManager for async operations.
"""Custom bookmark manager implementation:
from neo4j import BookmarkManager, GraphDatabase, basic_auth
class CustomBookmarkManager(BookmarkManager):
def __init__(self):
self._bookmarks = []
def get_bookmarks(self):
return self._bookmarks
def update_bookmarks(self, previous_bookmarks, new_bookmarks):
# Custom logic for bookmark management
self._bookmarks = new_bookmarks
# Could store in database, cache, etc.
# Use custom bookmark manager
custom_manager = CustomBookmarkManager()
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
with driver.session(bookmark_manager=custom_manager) as session:
session.run("CREATE (n:Person {name: 'Charlie'})")
# Bookmarks automatically managed by custom_managerAddress Classes
Network address handling for server connections and cluster routing.
class Address:
"""Base class for server addresses."""
@classmethod
def parse(
cls,
address: str,
default_host: str = None,
default_port: int = None
) -> Address:
"""
Parse address string to Address instance.
Parameters:
- address: Address string to parse
- default_host: Default host if not specified
- default_port: Default port if not specified
Returns:
Address instance (IPv4Address or IPv6Address)
"""
@classmethod
def parse_list(
cls,
addresses: str | list[str],
default_host: str = None,
default_port: int = None
) -> list[Address]:
"""
Parse multiple addresses.
Parameters:
- addresses: Address string or list of address strings
- default_host: Default host for addresses without host
- default_port: Default port for addresses without port
Returns:
List of Address instances
"""
class IPv4Address(Address):
"""
IPv4 address representation.
Contains host and port information.
"""
class IPv6Address(Address):
"""
IPv6 address representation.
Contains host, port, flow_info, and scope_id information.
"""Graph Data Types
Neo4j's rich graph type system including nodes, relationships, and paths returned from Cypher queries.
class Node:
"""
Self-contained graph node with labels and properties.
"""
@property
def element_id(self) -> str:
"""The unique identifier for this node."""
@property
def labels(self) -> frozenset[str]:
"""The set of labels attached to this node."""
def get(self, name: str, default: object = None) -> Any:
"""Get a property value by name, optionally with a default."""
def keys(self) -> KeysView[str]:
"""Return an iterable of all property names."""
def values(self) -> ValuesView[Any]:
"""Return an iterable of all property values."""
def items(self) -> ItemsView[str, Any]:
"""Return an iterable of all property name-value pairs."""
def __getitem__(self, name: str) -> Any:
"""Access property by name."""
def __contains__(self, name: object) -> bool:
"""Check if property exists."""
class Relationship:
"""
Self-contained graph relationship connecting two nodes.
"""
@property
def element_id(self) -> str:
"""The unique identifier for this relationship."""
@property
def start_node(self) -> Node | None:
"""Get the start node of this relationship."""
@property
def end_node(self) -> Node | None:
"""Get the end node of this relationship."""
@property
def type(self) -> str:
"""Get the type name of this relationship."""
@property
def nodes(self) -> tuple[Node | None, Node | None]:
"""Get the pair of nodes which this relationship connects."""
def get(self, name: str, default: object = None) -> Any:
"""Get a property value by name, optionally with a default."""
def keys(self) -> KeysView[str]:
"""Return an iterable of all property names."""
def values(self) -> ValuesView[Any]:
"""Return an iterable of all property values."""
def items(self) -> ItemsView[str, Any]:
"""Return an iterable of all property name-value pairs."""
class Path:
"""
Self-contained graph path representing a sequence of connected nodes and relationships.
"""
@property
def start_node(self) -> Node:
"""The first Node in this path."""
@property
def end_node(self) -> Node:
"""The last Node in this path."""
@property
def nodes(self) -> tuple[Node, ...]:
"""The sequence of Node objects in this path."""
@property
def relationships(self) -> tuple[Relationship, ...]:
"""The sequence of Relationship objects in this path."""
def __len__(self) -> int:
"""Get the number of relationships in the path."""
def __iter__(self) -> Iterator[Relationship]:
"""Iterate over relationships in the path."""Temporal Data Types
Neo4j's temporal types for date, time, and duration handling with nanosecond precision.
class Date:
"""
Idealized date representation with year, month, and day.
Years between 0001 and 9999 are supported.
"""
def __init__(self, year: int, month: int, day: int):
"""
Create a new Date.
Parameters:
- year: Year (1-9999)
- month: Month (1-12)
- day: Day of month (1-31)
"""
@property
def year(self) -> int:
"""The year of the date."""
@property
def month(self) -> int:
"""The month of the date."""
@property
def day(self) -> int:
"""The day of the date."""
@classmethod
def today(cls, tz: tzinfo | None = None) -> Date:
"""Get the current date."""
@classmethod
def from_iso_format(cls, s: str) -> Date:
"""Parse ISO formatted date string (YYYY-MM-DD)."""
def iso_format(self) -> str:
"""Return the Date as ISO formatted string."""
def to_native(self) -> date:
"""Convert to a native Python datetime.date value."""
class DateTime:
"""
A point in time with nanosecond precision, combining date and time components.
"""
def __init__(
self,
year: int,
month: int,
day: int,
hour: int = 0,
minute: int = 0,
second: int = 0,
nanosecond: int = 0,
tzinfo: tzinfo | None = None
):
"""
Create a new DateTime.
Parameters:
- year, month, day: Date components
- hour, minute, second, nanosecond: Time components
- tzinfo: Timezone information
"""
@property
def year(self) -> int:
"""The year of the DateTime."""
@property
def month(self) -> int:
"""The month of the DateTime."""
@property
def day(self) -> int:
"""The day of the DateTime."""
@property
def hour(self) -> int:
"""The hour of the DateTime."""
@property
def minute(self) -> int:
"""The minute of the DateTime."""
@property
def second(self) -> int:
"""The second of the DateTime."""
@property
def nanosecond(self) -> int:
"""The nanosecond of the DateTime."""
@property
def tzinfo(self) -> tzinfo | None:
"""The timezone information."""
@classmethod
def now(cls, tz: tzinfo | None = None) -> DateTime:
"""Get the current date and time."""
@classmethod
def from_iso_format(cls, s: str) -> DateTime:
"""Parse ISO formatted datetime string."""
def iso_format(self, sep: str = "T") -> str:
"""Return the DateTime as ISO formatted string."""
def to_native(self) -> datetime:
"""Convert to a native Python datetime.datetime value."""
class Time:
"""
Time of day with nanosecond precision.
"""
def __init__(
self,
hour: int = 0,
minute: int = 0,
second: int = 0,
nanosecond: int = 0,
tzinfo: tzinfo | None = None
):
"""
Create a new Time.
Parameters:
- hour: Hour (0-23)
- minute: Minute (0-59)
- second: Second (0-59)
- nanosecond: Nanosecond (0-999999999)
- tzinfo: Timezone information
"""
@property
def hour(self) -> int:
"""The hour of the time."""
@property
def minute(self) -> int:
"""The minute of the time."""
@property
def second(self) -> int:
"""The second of the time."""
@property
def nanosecond(self) -> int:
"""The nanosecond of the time."""
@property
def tzinfo(self) -> tzinfo | None:
"""The timezone information."""
@classmethod
def now(cls, tz: tzinfo | None = None) -> Time:
"""Get the current time."""
def to_native(self) -> time:
"""Convert to a native Python datetime.time value."""
class Duration:
"""
A difference between two points in time with months, days, seconds, and nanoseconds.
"""
def __init__(
self,
years: float = 0,
months: float = 0,
weeks: float = 0,
days: float = 0,
hours: float = 0,
minutes: float = 0,
seconds: float = 0,
milliseconds: float = 0,
microseconds: float = 0,
nanoseconds: float = 0
):
"""
Create a new Duration.
All parameters are optional and default to 0.
"""
@property
def months(self) -> int:
"""The months component of the Duration."""
@property
def days(self) -> int:
"""The days component of the Duration."""
@property
def seconds(self) -> int:
"""The seconds component of the Duration."""
@property
def nanoseconds(self) -> int:
"""The nanoseconds component of the Duration."""
@classmethod
def from_iso_format(cls, s: str) -> Duration:
"""Parse ISO formatted duration string."""
def iso_format(self, sep: str = "T") -> str:
"""Return the Duration as ISO formatted string."""Neo4j Type System Examples
Working with graph and temporal types in practice:
# Working with nodes and relationships
from neo4j import GraphDatabase, basic_auth
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
with driver.session() as session:
result = session.run("""
CREATE (a:Person {name: 'Alice', age: 30})
CREATE (b:Person {name: 'Bob', age: 25})
CREATE (a)-[r:KNOWS {since: date('2020-01-01')}]->(b)
RETURN a, b, r
""")
record = result.single()
# Access node properties
alice = record["a"]
print(f"Name: {alice['name']}, Age: {alice['age']}")
print(f"Labels: {alice.labels}")
print(f"Element ID: {alice.element_id}")
# Access relationship properties
relationship = record["r"]
print(f"Type: {relationship.type}")
print(f"Since: {relationship['since']}")
print(f"Start: {relationship.start_node.element_id}")
print(f"End: {relationship.end_node.element_id}")
# Working with temporal types
from neo4j.time import DateTime, Date, Time, Duration
with driver.session() as session:
# Neo4j temporal types
result = session.run("""
CREATE (e:Event {
date: date('2023-12-25'),
time: time('14:30:00'),
datetime: datetime('2023-12-25T14:30:00'),
duration: duration('P1Y2M3DT4H5M6S')
})
RETURN e
""")
event = result.single()["e"]
print(f"Date: {event['date']}")
print(f"Time: {event['time']}")
print(f"DateTime: {event['datetime']}")
print(f"Duration: {event['duration']}")
# Python datetime integration
python_datetime = datetime.now()
session.run(
"CREATE (e:Event {created: $dt})",
dt=python_datetime
)Spatial Data Types
Neo4j's spatial types for geometric data with coordinate reference systems (SRID).
class Point:
"""
Base class for spatial data representing a point in geometric space.
Contains coordinates and spatial reference identifier (SRID).
"""
def __init__(self, iterable: Iterable[float]):
"""
Create a Point from coordinates.
Parameters:
- iterable: Sequence of coordinate values as floats
"""
@property
def srid(self) -> int | None:
"""The spatial reference identifier for this point."""
@property
def x(self) -> float:
"""The x coordinate."""
@property
def y(self) -> float:
"""The y coordinate."""
@property
def z(self) -> float:
"""The z coordinate (if 3D)."""
class CartesianPoint(Point):
"""
Point in Cartesian coordinate system.
Uses x, y, z coordinates with SRIDs 7203 (2D) or 9157 (3D).
"""
@property
def x(self) -> float:
"""The x coordinate."""
@property
def y(self) -> float:
"""The y coordinate."""
@property
def z(self) -> float:
"""The z coordinate (3D only)."""
class WGS84Point(Point):
"""
Point in WGS84 geographic coordinate system (GPS coordinates).
Uses longitude, latitude, height with SRIDs 4326 (2D) or 4979 (3D).
"""
@property
def longitude(self) -> float:
"""The longitude coordinate."""
@property
def latitude(self) -> float:
"""The latitude coordinate."""
@property
def height(self) -> float:
"""The height coordinate (3D only)."""Exception Hierarchy
Neo4j driver exception hierarchy for error handling and troubleshooting.
class Neo4jError(Exception):
"""
Base class for all Neo4j related errors.
Represents errors returned by the Neo4j database.
"""
@property
def code(self) -> str:
"""The Neo4j error code."""
@property
def message(self) -> str:
"""The error message."""
class ClientError(Neo4jError):
"""
Client errors indicate problems with the request.
These are not retriable and indicate user errors.
"""
class DatabaseError(Neo4jError):
"""
Database errors indicate problems within the database.
These may be retriable depending on the specific error.
"""
class TransientError(Neo4jError):
"""
Transient errors are temporary and operations should be retried.
Indicates temporary issues that may resolve themselves.
"""
class DriverError(Exception):
"""
Base class for all driver-related errors.
Represents errors in the driver itself, not from the database.
"""
class ServiceUnavailable(DriverError):
"""
Raised when the driver cannot establish connection to any server.
This may indicate network issues or database unavailability.
"""
class SessionExpired(DriverError):
"""
Raised when a session is no longer valid.
Sessions can expire due to network issues or server restart.
"""
class TransactionError(DriverError):
"""
Base class for transaction-related errors.
"""
class ResultError(DriverError):
"""
Base class for result-related errors.
"""
class ResultConsumedError(ResultError):
"""
Raised when attempting to access an already consumed result.
"""
class ResultNotSingleError(ResultError):
"""
Raised when calling single() on a result with zero or multiple records.
"""
class AuthError(ClientError):
"""
Authentication errors indicate invalid credentials or auth issues.
"""
class CypherSyntaxError(ClientError):
"""
Raised for Cypher syntax errors in queries.
"""
class CypherTypeError(ClientError):
"""
Raised for Cypher type errors in queries.
"""
class ConstraintError(ClientError):
"""
Raised when a constraint violation occurs.
"""
class Forbidden(ClientError):
"""
Raised when access is denied to a resource or operation.
"""
class DatabaseUnavailable(TransientError):
"""
Raised when the requested database is unavailable.
"""
class NotALeader(TransientError):
"""
Raised when a write operation is attempted on a follower in a cluster.
"""Exception Handling Examples
Working with Neo4j exceptions for robust error handling:
from neo4j import GraphDatabase, basic_auth
from neo4j.exceptions import (
Neo4jError, ClientError, TransientError, DriverError,
ServiceUnavailable, AuthError, CypherSyntaxError,
ResultNotSingleError, SessionExpired
)
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
# Handle different error types appropriately
def run_query_with_error_handling():
try:
with driver.session() as session:
result = session.run("INVALID CYPHER QUERY")
return result.single()
except CypherSyntaxError as e:
print(f"Query syntax error: {e.message}")
# Fix the query and retry
except AuthError as e:
print(f"Authentication failed: {e.message}")
# Check credentials
except ServiceUnavailable as e:
print(f"Database unavailable: {e}")
# Implement retry logic or fallback
except TransientError as e:
print(f"Temporary error: {e.message}")
# Retry the operation
except ClientError as e:
print(f"Client error (do not retry): {e.code} - {e.message}")
# Fix the client-side issue
except Neo4jError as e:
print(f"Database error: {e.code} - {e.message}")
# Handle database-specific errors
except DriverError as e:
print(f"Driver error: {e}")
# Handle driver-related issues
# Retry logic for transient errors
import time
from random import uniform
def run_with_retry(session, query, max_retries=3):
for attempt in range(max_retries):
try:
return session.run(query)
except TransientError as e:
if attempt == max_retries - 1:
raise # Re-raise if last attempt
# Exponential backoff with jitter
wait_time = (2 ** attempt) + uniform(0, 1)
print(f"Transient error, retrying in {wait_time:.2f}s: {e.message}")
time.sleep(wait_time)
except (ClientError, DatabaseError):
# Don't retry client or database errors
raise
# Handle result errors
def get_single_user(session, name):
try:
result = session.run("MATCH (u:User {name: $name}) RETURN u", name=name)
return result.single()["u"]
except ResultNotSingleError as e:
if "no records" in str(e).lower():
print(f"User '{name}' not found")
return None
else:
print(f"Multiple users found with name '{name}'")
raiseSpatial Types Examples
# Working with spatial data
from neo4j import GraphDatabase, basic_auth
from neo4j.spatial import CartesianPoint, WGS84Point
driver = GraphDatabase.driver("bolt://localhost:7687", auth=basic_auth("neo4j", "password"))
with driver.session() as session:
result = session.run("""
CREATE (p:Place {
location: point({latitude: 37.7749, longitude: -122.4194}),
name: 'San Francisco'
})
RETURN p
""")
place = result.single()["p"]
location = place["location"] # WGS84Point instance
print(f"Coordinates: {location.latitude}, {location.longitude}")
print(f"SRID: {location.srid}")
# Create Cartesian point
cartesian = CartesianPoint([10.0, 20.0])
session.run(
"CREATE (p:Point {location: $point})",
point=cartesian
)Utility Functions
def unit_of_work(
metadata: dict | None = None,
timeout: float | None = None
) -> Callable:
"""
Decorator for transaction functions with metadata and timeout.
Parameters:
- metadata: Transaction metadata dictionary
- timeout: Transaction timeout in seconds
Returns:
Decorated function that passes metadata and timeout to transactions
"""Example decorator usage:
from neo4j import unit_of_work
@unit_of_work(metadata={"operation": "user_creation"}, timeout=30.0)
def create_user_with_friends(tx, user_name, friend_names):
# Create user
user_result = tx.run(
"CREATE (u:User {name: $name}) RETURN u",
name=user_name
)
user_id = user_result.single()["u"].id
# Create friends and relationships
for friend_name in friend_names:
tx.run("""
MATCH (u:User) WHERE id(u) = $user_id
MERGE (f:User {name: $friend_name})
CREATE (u)-[:FRIENDS_WITH]->(f)
""", user_id=user_id, friend_name=friend_name)
return user_id
# Use the decorated function
with driver.session() as session:
user_id = session.execute_write(
create_user_with_friends,
"Alice",
["Bob", "Charlie", "Diana"]
)
print(f"Created user with ID: {user_id}")Install with Tessl CLI
npx tessl i tessl/pypi-neo4j