🔍 Code Extractor

Validates that a Neo4j database schema is correctly configured by checking for required constraints, node labels, and indexes.

Purpose

This function performs comprehensive schema validation for a Neo4j graph database used in a controlled document management system (CDocs). It verifies the presence of essential database constraints (uniqueness constraints on UIDs), checks that required node types exist in the database, and ensures necessary indexes are created for performance optimization. The function is decorated with guard_execution to prevent excessive validation calls within a 5-second cooldown period. It returns a detailed report of any schema issues found, making it useful for database health checks, deployment validation, and troubleshooting.

Source Code

def validate_schema(driver: Driver) -> Tuple[bool, List[str]]:
    """
    Validate that the database schema is correctly set up.
            
    Args:
        driver: Neo4j driver instance
        
    Returns:
        Tuple of (success boolean, list of issues found)
    """
    logger.info("Beginning schema validation")
    issues = []
    
    try:
        with driver.session() as session:
            # Check for required constraints
            constraints = session.run("SHOW CONSTRAINTS").values()
            
            constraint_names = [row[0] for row in constraints]
            
            required_constraints = [
                "cdocs_uid_unique",
                "controlled_document_uid_unique",
                "document_version_uid_unique",
                "review_cycle_uid_unique",
                "reviewer_assignment_uid_unique",
                "review_comment_uid_unique",
                "approval_uid_unique",
                "approval_step_uid_unique",
                "approver_uid_unique",
                "approval_workflow_uid_unique",
                "document_type_code_unique",
                "department_code_unique",
                "user_uid_unique"
            ]
            
            for constraint in required_constraints:
                if constraint not in constraint_names:
                    issues.append(f"Missing constraint: {constraint}")
            
            # Check for required node types
            required_node_labels = [
                NodeLabels.CDOCS,
                NodeLabels.CONTROLLED_DOCUMENT,
                NodeLabels.DOCUMENT_VERSION,
                NodeLabels.REVIEW_CYCLE,
                NodeLabels.REVIEWER_ASSIGNMENT,
                NodeLabels.APPROVAL,
                NodeLabels.APPROVAL_STEP,
                NodeLabels.APPROVER,
                NodeLabels.USER
            ]
            
            for label in required_node_labels:
                result = session.run(
                    f"MATCH (n:{label}) RETURN count(n) > 0 as exists"
                )
                record = result.single()
                if not record or not record["exists"]:
                    issues.append(f"No nodes found with label: {label}")
            
            # Check for required indexes
            indexes = session.run("SHOW INDEXES").values()
            index_descriptions = [row[8] for row in indexes if len(row) > 8]
            
            required_indexes = [
                "cdocs_doc_number_idx",
                "document_version_number_idx",
                "document_status_idx",
                "review_status_idx",
                "approval_status_idx",
                "reviewer_uid_idx",
                "approver_uid_idx"
            ]
            
            for index in required_indexes:
                if not any(index in desc for desc in index_descriptions):
                    issues.append(f"Missing index: {index}")
            
            if issues:
                logger.warning(f"Schema validation found issues: {issues}")
                return False, issues
            else:
                logger.info("Schema validation passed")
                return True, []
                
    except Exception as e:
        logger.error(f"Schema validation failed with error: {e}")
        logger.error(traceback.format_exc())
        issues.append(f"Validation error: {str(e)}")
        return False, issues

Parameters

Parameter Details

driver: A Neo4j Driver instance that provides the connection to the Neo4j database. This should be an active, authenticated driver object created using neo4j.GraphDatabase.driver(). The driver is used to create sessions for executing Cypher queries to inspect the database schema.

Return Value

Type: Tuple[bool, List[str]]

Returns a Tuple[bool, List[str]] where the first element is a boolean indicating validation success (True if all checks pass, False if any issues found), and the second element is a list of strings describing any issues discovered. If validation passes, the list will be empty. Issue strings follow patterns like 'Missing constraint: {name}', 'No nodes found with label: {label}', or 'Missing index: {name}'.

Dependencies

• neo4j
• logging
• uuid
• traceback
• typing
• datetime
• CDocs

Required Imports

from neo4j import Driver
from typing import Tuple, List
import logging
import traceback
from CDocs import guard_execution

Usage Example

from neo4j import GraphDatabase
from typing import Tuple, List
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create Neo4j driver
driver = GraphDatabase.driver(
    'bolt://localhost:7687',
    auth=('neo4j', 'password')
)

try:
    # Validate the schema
    is_valid, issues = validate_schema(driver)
    
    if is_valid:
        print('Schema validation passed!')
    else:
        print('Schema validation failed with issues:')
        for issue in issues:
            print(f'  - {issue}')
finally:
    driver.close()

Best Practices

• Ensure the Neo4j driver is properly initialized and authenticated before calling this function
• The function is protected by a 5-second cooldown via guard_execution decorator, so avoid calling it in tight loops
• Review the returned issues list carefully as it provides specific details about missing schema elements
• Run this validation after database migrations or initial setup to ensure schema integrity
• The function expects specific NodeLabels constants to be defined in the CDocs module - ensure these match your schema design
• Consider running this validation as part of application startup or health check endpoints
• The function checks for node existence, not just schema definitions - ensure at least one node of each type exists for full validation
• Handle the returned tuple appropriately - the boolean indicates overall success, but the issues list provides actionable details
• Be aware that this function performs multiple database queries and may take time on large databases

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function validate_schema_v1 97.1% similar

  Validates that a Neo4j database schema contains all required constraints and node labels for a controlled document management system.

  From: /tf/active/vicechatdev/CDocs/db/schema_manager.py

• function initialize_schema 77.5% similar

  Initializes the Neo4j database schema by creating required constraints, indexes, root nodes, audit trails, and migrating existing data structures to current schema versions.

  From: /tf/active/vicechatdev/CDocs/db/schema_manager.py

• function initialize_schema_v1 76.2% similar

  Initializes a Neo4j database schema by creating constraints, indexes, a root CDocs node, audit trail, and migrating approval data and workflow types.

  From: /tf/active/vicechatdev/CDocs single class/db/schema_manager.py

• function update_schema 63.2% similar

  Updates a Neo4j database schema to match a specific version by running base schema initialization and applying version-specific migrations sequentially.

  From: /tf/active/vicechatdev/CDocs single class/db/schema_manager.py

• function init_database 61.2% similar

  Initializes a Neo4j database with required schema constraints, creates an AuditTrail node, and migrates existing audit events.

  From: /tf/active/vicechatdev/CDocs/db/__init__.py