tessl/pypi-neo4j
Neo4j Bolt driver for Python providing database connectivity and query execution
—
Pending
drivers.mddocs/
Drivers & Connection
Connection management and driver creation for Neo4j databases, supporting both direct server connections and cluster routing with comprehensive configuration options. The driver handles connection pooling, authentication, encryption, and provides both synchronous and asynchronous APIs.
Capabilities
GraphDatabase Factory
Static factory class providing convenient methods for creating different types of drivers based on connection requirements and database topology.
class GraphDatabase:
@staticmethod
def driver(uri: str, *, auth: Auth = None, **config) -> Driver:
"""
Create a driver instance for connecting to Neo4j.
Parameters:
- uri: Connection URI (bolt://, neo4j://, bolt+s://, neo4j+s://)
- auth: Authentication token (basic_auth, kerberos_auth, etc.)
- **config: Driver configuration options
Returns:
Driver instance (BoltDriver or Neo4jDriver based on URI scheme)
"""
@staticmethod
def bookmark_manager(
initial_bookmarks: list[Bookmark] = None,
bookmarks_supplier: Callable = None,
bookmarks_consumer: Callable = None
) -> BookmarkManager:
"""
Create a default BookmarkManager implementation.
Parameters:
- initial_bookmarks: Starting bookmarks for causal consistency
- bookmarks_supplier: Function to supply bookmarks
- bookmarks_consumer: Function to consume bookmarks
Returns:
BookmarkManager instance for managing causal consistency
"""
@staticmethod
def bolt_driver(target: str, **config) -> BoltDriver:
"""
Create a direct Bolt driver for single server connections.
Parameters:
- target: Bolt server address
- **config: Driver configuration options
Returns:
BoltDriver instance for direct server connection
"""
@staticmethod
def neo4j_driver(
*targets: str,
routing_context: dict = None,
**config
) -> Neo4jDriver:
"""
Create a routing driver for Neo4j cluster connections.
Parameters:
- *targets: Cluster member addresses
- routing_context: Routing context for cluster discovery
- **config: Driver configuration options
Returns:
Neo4jDriver instance for cluster routing
"""Example usage:
from neo4j import GraphDatabase, basic_auth
# Create driver with automatic type selection
driver = GraphDatabase.driver(
"neo4j://localhost:7687",
auth=basic_auth("neo4j", "password"),
max_connection_lifetime=30 * 60, # 30 minutes
max_connection_pool_size=50,
connection_acquisition_timeout=60.0
)
# Create direct bolt driver
bolt_driver = GraphDatabase.bolt_driver(
"bolt://localhost:7687",
auth=basic_auth("neo4j", "password")
)
# Create cluster routing driver
cluster_driver = GraphDatabase.neo4j_driver(
"localhost:7687",
"localhost:7688",
"localhost:7689",
routing_context={"region": "us-east-1"}
)AsyncGraphDatabase Factory
Asynchronous version of GraphDatabase providing the same factory methods but returning async-compatible drivers.
class AsyncGraphDatabase:
@staticmethod
def driver(uri: str, *, auth: Auth = None, **config) -> AsyncDriver:
"""
Create an async driver instance for connecting to Neo4j.
Returns:
AsyncDriver instance with async methods
"""
@staticmethod
def bookmark_manager(
initial_bookmarks: list[Bookmark] = None,
bookmarks_supplier: Callable = None,
bookmarks_consumer: Callable = None
) -> AsyncBookmarkManager:
"""
Create an async BookmarkManager implementation.
Returns:
AsyncBookmarkManager for async bookmark management
"""
@staticmethod
def bolt_driver(target: str, **config) -> AsyncBoltDriver:
"""Create async direct Bolt driver."""
@staticmethod
def neo4j_driver(
*targets: str,
routing_context: dict = None,
**config
) -> AsyncNeo4jDriver:
"""Create async routing driver."""Driver Base Class
Base class for all synchronous driver types providing core functionality for session creation, query execution, and connection management.
class Driver:
@property
def encrypted(self) -> bool:
"""Indicates if driver uses encryption."""
@property
def execute_query_bookmark_manager(self) -> BookmarkManager:
"""Default bookmark manager for execute_query operations."""
def session(
self,
*,
database: str = None,
default_access_mode: str = None,
bookmarks: list[Bookmark] = None,
bookmark_manager: BookmarkManager = None,
auth: Auth = None,
**config
) -> Session:
"""
Create a new session for executing work.
Parameters:
- database: Target database name (None for default)
- default_access_mode: READ_ACCESS or WRITE_ACCESS
- bookmarks: Initial bookmarks for causal consistency
- bookmark_manager: Custom bookmark manager
- auth: Session-level authentication override
- **config: Additional session configuration
Returns:
Session instance for executing queries and transactions
"""
def execute_query(
self,
query: str,
parameters: dict = None,
routing: RoutingControl = RoutingControl.WRITE,
database: str = None,
impersonated_user: str = None,
bookmark_manager: BookmarkManager = None,
auth: Auth = None,
result_transformer: Callable = None,
**kwargs
) -> Any:
"""
Execute a query in a managed transaction function.
Parameters:
- query: Cypher query string
- parameters: Query parameters dictionary
- routing: Query routing control (READ/WRITE)
- database: Target database name
- impersonated_user: User to impersonate
- bookmark_manager: Bookmark manager for causal consistency
- auth: Authentication override
- result_transformer: Function to transform results
Returns:
Transformed query results (default: records, summary, keys tuple)
"""
def close(self) -> None:
"""
Shut down the driver and close all connections.
Blocks until all connections are closed.
"""
def verify_connectivity(
self,
*,
database: str = None,
auth: Auth = None
) -> None:
"""
Verify that the driver can connect to the server.
Parameters:
- database: Database to test connectivity against
- auth: Authentication credentials to test
Raises:
Exception if connectivity verification fails
"""
def get_server_info(
self,
*,
database: str = None,
auth: Auth = None
) -> ServerInfo:
"""
Get information about the connected Neo4j server.
Parameters:
- database: Target database for server info
- auth: Authentication credentials
Returns:
ServerInfo containing server details
"""
def verify_authentication(
self,
*,
auth: Auth = None,
**config
) -> bool:
"""
Verify authentication credentials against the server.
Parameters:
- auth: Authentication credentials to verify
- **config: Additional verification configuration
Returns:
True if authentication is valid, False otherwise
"""
def supports_multi_db(self) -> bool:
"""
Check if the server supports multiple databases.
Returns:
True if multi-database is supported
"""
def supports_session_auth(self) -> bool:
"""
Check if the server supports session-level re-authentication.
Returns:
True if session auth is supported
"""AsyncDriver Base Class
Asynchronous version of the Driver base class with async methods for all operations.
class AsyncDriver:
@property
def encrypted(self) -> bool:
"""Indicates if driver uses encryption."""
@property
def execute_query_bookmark_manager(self) -> AsyncBookmarkManager:
"""Default async bookmark manager for execute_query operations."""
def session(
self,
*,
database: str = None,
default_access_mode: str = None,
bookmarks: list[Bookmark] = None,
bookmark_manager: AsyncBookmarkManager = None,
auth: Auth = None,
**config
) -> AsyncSession:
"""Create async session."""
async def execute_query(
self,
query: str,
parameters: dict = None,
routing: RoutingControl = RoutingControl.WRITE,
**kwargs
) -> Any:
"""Execute query asynchronously."""
async def close(self) -> None:
"""Asynchronously close driver and connections."""
async def verify_connectivity(self, **config) -> None:
"""Verify connectivity asynchronously."""
async def get_server_info(self, **config) -> ServerInfo:
"""Get server info asynchronously."""
async def verify_authentication(self, **config) -> bool:
"""Verify authentication asynchronously."""
async def supports_multi_db(self) -> bool:
"""Check multi-database support asynchronously."""
async def supports_session_auth(self) -> bool:
"""Check session auth support asynchronously."""Specific Driver Types
Concrete driver implementations for different connection patterns.
class BoltDriver(Driver):
"""
Direct driver for single Bolt server connections.
Inherits all Driver methods and properties.
"""
class AsyncBoltDriver(AsyncDriver):
"""
Async direct driver for single Bolt server connections.
Inherits all AsyncDriver methods and properties.
"""
class Neo4jDriver(Driver):
"""
Routing driver for Neo4j cluster connections.
Inherits all Driver methods with cluster routing capabilities.
"""
class AsyncNeo4jDriver(AsyncDriver):
"""
Async routing driver for Neo4j cluster connections.
Inherits all AsyncDriver methods with cluster routing capabilities.
"""Usage Examples
Basic Driver Creation
from neo4j import GraphDatabase, basic_auth
# Simple driver creation
driver = GraphDatabase.driver(
"bolt://localhost:7687",
auth=basic_auth("neo4j", "password")
)
# Driver with configuration
driver = GraphDatabase.driver(
"neo4j://cluster.example.com:7687",
auth=basic_auth("neo4j", "password"),
max_connection_lifetime=30 * 60,
max_connection_pool_size=50,
connection_acquisition_timeout=60.0,
resolver=custom_resolver_function
)Async Driver Usage
import asyncio
from neo4j import AsyncGraphDatabase, basic_auth
async def example():
driver = AsyncGraphDatabase.driver(
"neo4j://localhost:7687",
auth=basic_auth("neo4j", "password")
)
# Use execute_query for simple operations
records, summary, keys = await driver.execute_query(
"MATCH (n:Person) RETURN n.name AS name LIMIT 10"
)
await driver.close()
asyncio.run(example())Cluster Configuration
from neo4j import GraphDatabase, basic_auth
# Create cluster driver with multiple seed addresses
driver = GraphDatabase.neo4j_driver(
"server1.cluster.com:7687",
"server2.cluster.com:7687",
"server3.cluster.com:7687",
auth=basic_auth("neo4j", "password"),
routing_context={
"region": "us-west-2",
"zone": "us-west-2a"
}
)Types
class ServerInfo:
"""Information about the connected Neo4j server."""
address: Address
protocol_version: tuple
agent: str
class BookmarkManager:
"""Abstract base class for managing bookmarks."""
class AsyncBookmarkManager:
"""Abstract base class for async bookmark management."""
class RoutingControl:
"""Enum for controlling query routing."""
READ: str
WRITE: strInstall with Tessl CLI
npx tessl i tessl/pypi-neo4j