🔍 Code Extractor

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

Purpose

This function sets up the complete database schema for a CDocs (document management) system in Neo4j. It executes predefined schema queries, ensures a root node exists, creates an audit trail structure, migrates existing audit events and approval data to new structures, and initializes approval workflow types. It's designed to be idempotent and safe to run multiple times, with a cooldown guard to prevent excessive execution.

Source Code

def initialize_schema(driver: Driver) -> bool:
    """
    Initialize Neo4j schema with required constraints and indexes.
    
    Args:
        driver: Neo4j driver instance
        
    Returns:
        True if successful, False otherwise
    """
    try:
        # Execute schema queries
        with driver.session() as session:
            for query in SCHEMA_INITIALIZATION_QUERIES:
                try:
                    session.run(query)
                except Exception as e:
                    logger.error(f"Error executing schema query: {e}")
                    
            # Create root node if it doesn't exist
            result = session.run("MATCH (c:CDocs) RETURN c LIMIT 1")
            if not result.single():
                session.run("""
                    CREATE (c:CDocs {
                        UID: $uid,
                        name: 'CDocs Root',
                        version: '1.0.0',
                        created_at: datetime()
                    })
                """, {"uid": str(uuid.uuid4())})
                logger.info("Created CDocs root node")
            
            # Create audit trail if it doesn't exist and migrate events
            audit_trail_uid = ensure_audit_trail_exists(driver)
            if audit_trail_uid:
                # Pass the audit_trail_uid parameter to migrate_audit_events
                migrate_audit_events(driver=driver, audit_trail_uid=audit_trail_uid)
            else:
                logger.warning("Could not create or find audit trail - skipping event migration")
            
            # Migrate approval data to new structure if needed
            approval_migration_result = migrate_approval_data(driver)
            logger.info(f"Approval migration result: {approval_migration_result}")
            
            # Initialize workflow types
            initialize_approval_workflow_types(driver)
            
            logger.info("Neo4j schema initialized successfully")
            return True
    except Exception as e:
        logger.error(f"Error initializing Neo4j schema: {e}")
        logger.error(traceback.format_exc())
        return False

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 and managing transactions.

Return Value

Type: bool

Returns a boolean value: True if the schema initialization completed successfully (all queries executed, root node created/verified, audit trail established, and migrations completed), False if any exception occurred during the process. Note that individual query failures within the schema initialization loop are logged but don't immediately halt execution.

Dependencies

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

Required Imports

import logging
import uuid
import traceback
from typing import Dict, List, Any, Optional, Tuple
from neo4j import Driver, Session, Transaction
from datetime import datetime
from CDocs import guard_execution, db

Usage Example

from neo4j import GraphDatabase
from CDocs import guard_execution
import logging

# Configure logging
logger = logging.getLogger(__name__)

# Define schema queries (example)
SCHEMA_INITIALIZATION_QUERIES = [
    "CREATE CONSTRAINT IF NOT EXISTS FOR (c:CDocs) REQUIRE c.UID IS UNIQUE",
    "CREATE INDEX IF NOT EXISTS FOR (c:CDocs) ON (c.name)"
]

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

try:
    # Initialize schema
    success = initialize_schema(driver)
    if success:
        print("Schema initialized successfully")
    else:
        print("Schema initialization failed")
finally:
    driver.close()

Best Practices

• This function is decorated with guard_execution(cooldown_ms=5000), meaning it can only be executed once every 5 seconds to prevent excessive schema operations
• The function is designed to be idempotent - safe to run multiple times without causing errors or duplicate data
• Always ensure the Neo4j driver is properly closed after use to prevent connection leaks
• Individual schema query failures are logged but don't stop execution, allowing partial schema setup in case of errors
• The function requires several helper functions (ensure_audit_trail_exists, migrate_audit_events, migrate_approval_data, initialize_approval_workflow_types) to be defined in the same module or imported
• SCHEMA_INITIALIZATION_QUERIES should be defined as a module-level constant containing all necessary Cypher queries for constraints and indexes
• Monitor logs for warnings about audit trail creation failures, as this will skip event migration
• The function creates a root CDocs node with a unique UID - ensure this doesn't conflict with existing application logic
• Consider running this function during application startup or as part of a deployment script rather than on every request

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function initialize_schema 94.7% 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 init_database 82.9% 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

• function validate_schema 76.2% similar

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

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

• function validate_schema_v1 73.6% 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 update_schema 71.8% 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