tessl/pypi-neo4j

Neo4j Bolt driver for Python providing database connectivity and query execution

—

Pending

Overview

Eval results

Files

Configuration

Name: tessl/pypi-neo4j
Author: tessl

Configuration management including SSL/TLS trust settings, connection pooling, timeout configuration, and cluster routing options. The configuration system provides comprehensive control over driver behavior, security settings, and performance tuning.

Capabilities

Trust Store Configuration

SSL/TLS certificate trust configuration for secure connections to Neo4j servers.

class TrustSystemCAs:
    """
    Trust system certificate authorities (default behavior).
    Uses the operating system's certificate store to validate server certificates.
    """

class TrustAll:
    """
    Trust all certificates (insecure - for development only).
    Accepts any server certificate without validation.
    WARNING: Should never be used in production environments.
    """

class TrustCustomCAs:
    """
    Trust custom certificate authorities.
    Allows specification of custom certificate files for validation.
    """
    def __init__(self, *certs: str):
        """
        Initialize with custom certificate files.
        
        Parameters:
        - *certs: Paths to certificate files to trust
        """

Trust configuration examples:

from neo4j import GraphDatabase, TrustSystemCAs, TrustAll, TrustCustomCAs, basic_auth

# Use system certificate authorities (default)
driver = GraphDatabase.driver(
    "bolt+s://secure.neo4j.com:7687",
    auth=basic_auth("neo4j", "password"),
    trust=TrustSystemCAs()
)

# Development only - trust all certificates (insecure)
driver = GraphDatabase.driver(
    "bolt+s://localhost:7687", 
    auth=basic_auth("neo4j", "password"),
    trust=TrustAll()
)

# Trust custom certificate authorities
driver = GraphDatabase.driver(
    "bolt+s://internal.company.com:7687",
    auth=basic_auth("neo4j", "password"), 
    trust=TrustCustomCAs("/path/to/company-ca.pem", "/path/to/root-ca.pem")
)

# Using trust constants
import neo4j

driver = GraphDatabase.driver(
    "bolt+s://neo4j.com:7687",
    auth=basic_auth("neo4j", "password"),
    trust=neo4j.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES
)

Connection Pool Configuration

Driver configuration for connection pooling, timeouts, and connection management.

from neo4j import GraphDatabase, basic_auth

driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password"),
    
    # Connection pool settings
    max_connection_pool_size=50,           # Maximum connections in pool
    max_connection_lifetime=30 * 60,       # 30 minutes in seconds
    connection_acquisition_timeout=60.0,    # 60 seconds
    
    # Socket settings
    connection_timeout=30.0,               # Connection establishment timeout
    max_transaction_retry_time=30.0,       # Transaction retry timeout
    
    # Encryption settings
    encrypted=True,                        # Force encryption
    trust=neo4j.TRUST_SYSTEM_CA_SIGNED_CERTIFICATES,
    
    # User agent customization
    user_agent="MyApplication/1.0",
    
    # Resolver for custom address resolution
    resolver=custom_resolver_function,
    
    # Keep alive settings
    keep_alive=True,
    
    # Update routing table settings
    update_routing_table_timeout=30.0
)

Notification Configuration

Configuration for controlling server notifications and their severity levels.

class NotificationMinimumSeverity:
    """Enum for minimum notification severity levels."""
    OFF: str          # Disable all notifications
    WARNING: str      # Only warning and higher severity
    INFORMATION: str  # All notifications (default)

class NotificationSeverity:
    """Enum for notification severity levels."""
    WARNING: str      # Warning severity
    INFORMATION: str  # Information severity
    UNKNOWN: str      # Unknown severity

class NotificationCategory:
    """Enum for notification categories."""
    # Various notification categories for different types of server messages

Notification configuration example:

from neo4j import GraphDatabase, NotificationMinimumSeverity, basic_auth

# Configure notifications
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password"),
    notifications_min_severity=NotificationMinimumSeverity.WARNING,
    notifications_disabled_categories=[
        "UNSUPPORTED",  # Disable unsupported feature warnings
        "PERFORMANCE"   # Disable performance warnings
    ]
)

# Check notifications in results
with driver.session() as session:
    result = session.run("MATCH (n) RETURN n")
    summary = result.consume()
    
    for notification in summary.notifications:
        print(f"Severity: {notification.severity}")
        print(f"Category: {notification.category}")
        print(f"Title: {notification.title}")
        print(f"Description: {notification.description}")

Routing Configuration

Configuration for cluster routing and load balancing in Neo4j clusters.

class RoutingControl:
    """Enum for controlling query routing in clusters."""
    READ: str   # Route to read replicas
    WRITE: str  # Route to write servers (default)

Cluster routing configuration:

from neo4j import GraphDatabase, RoutingControl, basic_auth

# Cluster driver with routing configuration
driver = GraphDatabase.neo4j_driver(
    "server1.cluster.com:7687",
    "server2.cluster.com:7687", 
    "server3.cluster.com:7687",
    auth=basic_auth("neo4j", "password"),
    
    # Routing context for cluster discovery
    routing_context={
        "region": "us-west-2",
        "zone": "us-west-2a",
        "policy": "datacenter_local"
    },
    
    # Routing table refresh settings
    max_connection_lifetime=3600,          # 1 hour
    update_routing_table_timeout=30.0      # 30 seconds
)

# Execute queries with routing control
with driver.session() as session:
    # Write operation (routed to write servers)
    session.execute_write(lambda tx: tx.run("CREATE (n:Person {name: 'Alice'})"))
    
    # Read operation (routed to read replicas)
    def read_people(tx):
        return tx.run("MATCH (n:Person) RETURN n.name AS name").data()
    
    people = session.execute_read(read_people)

Database Configuration

Configuration for multi-database support and database selection.

from neo4j import GraphDatabase, basic_auth

# Default database constants
import neo4j
print(neo4j.DEFAULT_DATABASE)  # None (uses server default)
print(neo4j.SYSTEM_DATABASE)   # "system"

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

# Sessions with specific databases
with driver.session(database="myapp") as session:
    session.run("CREATE (n:User {name: 'Alice'})")

with driver.session(database="analytics") as session:
    session.run("CREATE (n:Event {type: 'login'})")

with driver.session(database=neo4j.SYSTEM_DATABASE) as session:
    result = session.run("SHOW DATABASES")
    for record in result:
        print(record["name"])

Access Mode Configuration

Configuration for read/write access control and session behavior.

# Access mode constants
READ_ACCESS: str = "READ"
WRITE_ACCESS: str = "WRITE"

Access mode configuration:

from neo4j import GraphDatabase, READ_ACCESS, WRITE_ACCESS, basic_auth

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

# Read-only session
with driver.session(default_access_mode=READ_ACCESS) as session:
    # Only read operations allowed
    result = session.run("MATCH (n:Person) RETURN n.name")
    
# Write session (default)
with driver.session(default_access_mode=WRITE_ACCESS) as session:
    session.run("CREATE (n:Person {name: 'Bob'})")

Advanced Configuration Options

Comprehensive driver configuration with all available options:

from neo4j import GraphDatabase, basic_auth, TrustSystemCAs

def custom_resolver(address):
    """Custom address resolver for service discovery."""
    # Implement custom DNS resolution logic
    return resolved_addresses

driver = GraphDatabase.driver(
    "neo4j://cluster.example.com:7687",
    auth=basic_auth("neo4j", "password"),
    
    # Connection settings
    max_connection_pool_size=100,
    max_connection_lifetime=1800,           # 30 minutes
    connection_acquisition_timeout=120.0,   # 2 minutes
    connection_timeout=30.0,
    
    # Transaction settings
    max_transaction_retry_time=30.0,
    default_access_mode=WRITE_ACCESS,
    
    # Security settings
    encrypted=True,
    trust=TrustSystemCAs(),
    
    # Notification settings
    notifications_min_severity="WARNING",
    notifications_disabled_categories=["PERFORMANCE"],
    
    # Routing settings (for cluster drivers)
    routing_context={"region": "us-east-1"},
    update_routing_table_timeout=30.0,
    
    # Custom settings
    user_agent="MyApp/2.0 (Production)",
    resolver=custom_resolver,
    keep_alive=True,
    
    # Session defaults
    database="production",
    bookmarks=None,
    
    # Logging and debugging
    # log_level="DEBUG"  # Not directly available, use Python logging
)

Environment-Based Configuration

Configuration using environment variables and configuration files:

import os
from neo4j import GraphDatabase, basic_auth, TrustSystemCAs

class Neo4jConfig:
    def __init__(self):
        # Connection settings from environment
        self.uri = os.getenv("NEO4J_URI", "bolt://localhost:7687")
        self.username = os.getenv("NEO4J_USERNAME", "neo4j") 
        self.password = os.getenv("NEO4J_PASSWORD")
        self.database = os.getenv("NEO4J_DATABASE", None)
        
        # Pool settings
        self.max_pool_size = int(os.getenv("NEO4J_MAX_POOL_SIZE", "50"))
        self.max_lifetime = int(os.getenv("NEO4J_MAX_LIFETIME", "3600"))
        
        # Security settings
        self.encrypted = os.getenv("NEO4J_ENCRYPTED", "true").lower() == "true"
        self.trust_level = os.getenv("NEO4J_TRUST", "SYSTEM_CA")
        
        if not self.password:
            raise ValueError("NEO4J_PASSWORD environment variable required")
    
    def create_driver(self):
        trust_config = TrustSystemCAs()  # Default to system CAs
        
        return GraphDatabase.driver(
            self.uri,
            auth=basic_auth(self.username, self.password),
            max_connection_pool_size=self.max_pool_size,
            max_connection_lifetime=self.max_lifetime,
            encrypted=self.encrypted,
            trust=trust_config
        )

# Usage
config = Neo4jConfig()
driver = config.create_driver()

Performance Tuning

Configuration for optimal performance in different scenarios:

from neo4j import GraphDatabase, basic_auth

# High-throughput configuration
high_throughput_driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password"),
    max_connection_pool_size=200,          # Large pool for concurrent requests
    connection_acquisition_timeout=5.0,     # Fast timeout to avoid blocking
    max_connection_lifetime=300,           # Short lifetime for refresh
    max_transaction_retry_time=5.0         # Quick retry for high throughput
)

# Long-running batch processing configuration
batch_driver = GraphDatabase.driver(
    "bolt://localhost:7687", 
    auth=basic_auth("neo4j", "password"),
    max_connection_pool_size=10,           # Fewer connections for batch jobs
    connection_acquisition_timeout=300.0,   # Longer timeout for availability
    max_connection_lifetime=7200,          # 2 hours for stable connections  
    max_transaction_retry_time=300.0       # Longer retry for batch operations
)

# Development/testing configuration
dev_driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password"),
    max_connection_pool_size=5,            # Small pool for development
    connection_acquisition_timeout=30.0,
    max_connection_lifetime=600,           # 10 minutes
    encrypted=False                        # Disable encryption for local dev
)

Configuration Constants

# Trust configuration constants
TRUST_SYSTEM_CA_SIGNED_CERTIFICATES: str = "TRUST_SYSTEM_CA_SIGNED_CERTIFICATES"
TRUST_ALL_CERTIFICATES: str = "TRUST_ALL_CERTIFICATES"

# Access mode constants  
READ_ACCESS: str = "READ"
WRITE_ACCESS: str = "WRITE"

# Database constants
DEFAULT_DATABASE: None = None
SYSTEM_DATABASE: str = "system"

Utility Functions

def get_user_agent() -> str:
    """
    Get the driver's default user agent string.
    
    Returns:
    Default user agent string sent to the server
    """

__version__: str
"""Driver version string."""

Example utility usage:

import neo4j

print(f"Neo4j Python Driver version: {neo4j.__version__}")
print(f"User agent: {neo4j.get_user_agent()}")

# Custom user agent
custom_agent = f"MyApp/1.0 {neo4j.get_user_agent()}"
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=basic_auth("neo4j", "password"),
    user_agent=custom_agent
)

Install with Tessl CLI