🔍 Code Extractor

Migrates legacy single-step approval records in Neo4j to a new multi-step approval model by creating ApprovalStep nodes and Approver nodes with proper relationships.

Purpose

This function performs a database migration to transform old-style Approval nodes (without HAS_STEP relationships) into the new multi-step approval architecture. It creates a default ApprovalStep for each old approval, migrates approver relationships, and preserves historical data including status, dates, and user information. The migration is batched (1000 records at a time) and includes a pre-check to avoid unnecessary operations. This is typically used during system upgrades or schema changes to maintain backward compatibility while adopting a more flexible approval workflow system.

Source Code

def migrate_approval_data(driver: Driver) -> Dict[str, Any]:
    """
    Migrate approval data to the new multi-step approval model.
    
    Args:
        driver: Neo4j driver instance
        
    Returns:
        Dict containing migration results
    """
    try:
        with driver.session() as session:
            # Check for old-style approvals that need migration
            result = session.run(
                """
                MATCH (a:Approval)
                WHERE NOT EXISTS((a)-[:HAS_STEP]->(:ApprovalStep))
                RETURN count(a) as count
                """
            )
            record = result.single()
            
            if not record or record["count"] == 0:
                return {
                    "success": True,
                    "message": "No approvals need migration",
                    "migrated_count": 0
                }
            
            # Perform migration for each old approval
            migration_query = """
            MATCH (a:Approval)
            WHERE NOT EXISTS((a)-[:HAS_STEP]->(:ApprovalStep))
            WITH a LIMIT 1000
            
            // Create a default step for each old approval
            CREATE (s:ApprovalStep {
                UID: apoc.create.uuid(),
                status: a.status,
                sequence_order: 1,
                step_number: 1,
                created_date: a.startDate,
                completion_date: a.completionDate,
                all_must_approve: true,
                required_approvals: 1
            })
            CREATE (a)-[:HAS_STEP]->(s)
            
            // Migrate approvers
            WITH a, s
            MATCH (a)-[r:APPROVED_BY]->(u:User)
            CREATE (apr:Approver {
                UID: apoc.create.uuid(),
                approver_uid: u.UID,
                user_uid: u.UID,
                approver_name: u.Name,
                user_name: u.Name,
                status: 'COMPLETED',
                role: 'Approver',
                assigned_date: a.startDate,
                assigned_at: a.startDate,
                decision: CASE WHEN a.status = 'APPROVED' THEN 'APPROVED' ELSE 'REJECTED' END,
                decision_date: a.completionDate
            })
            CREATE (s)-[:HAS_APPROVER]->(apr)
            
            RETURN count(a) as migrated
            """
            
            result = session.run(migration_query)
            record = result.single()
            
            return {
                "success": True,
                "message": f"Successfully migrated {record['migrated']} approvals",
                "migrated_count": record["migrated"]
            }
            
    except Exception as e:
        logger.error(f"Error migrating approval data: {e}")
        logger.error(traceback.format_exc())
        return {
            "success": False,
            "message": f"Error: {str(e)}",
            "migrated_count": 0
        }

Parameters

Parameter Details

driver: A Neo4j Driver instance used to establish database connections and execute Cypher queries. Must be a valid, connected Neo4j driver object with appropriate credentials and permissions to read and write Approval, ApprovalStep, Approver, and User nodes.

Return Value

Type: Dict[str, Any]

Returns a dictionary with three keys: 'success' (boolean indicating if migration completed without errors), 'message' (string describing the outcome or error details), and 'migrated_count' (integer representing the number of approval records successfully migrated). On success with no records to migrate, migrated_count is 0. On error, success is False and message contains the exception details.

Dependencies

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

Required Imports

from neo4j import Driver
from typing import Dict, Any
import logging
import traceback
from CDocs import guard_execution

Usage Example

from neo4j import GraphDatabase
from typing import Dict, Any
import logging

# Setup logger
logger = logging.getLogger(__name__)

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

# Execute migration
result = migrate_approval_data(driver)

if result["success"]:
    print(f"Migration completed: {result['message']}")
    print(f"Records migrated: {result['migrated_count']}")
else:
    print(f"Migration failed: {result['message']}")

# Close driver when done
driver.close()

Best Practices

• Ensure APOC plugin is installed in Neo4j before running this migration as it uses apoc.create.uuid()
• The function is decorated with guard_execution(cooldown_ms=5000) to prevent rapid repeated executions
• Migration processes 1000 records at a time (LIMIT 1000) to avoid memory issues with large datasets
• Always backup the Neo4j database before running migration functions
• The function is idempotent - it checks for approvals without HAS_STEP relationships before migrating
• Monitor the logger output for detailed error information if migration fails
• Consider running this during low-traffic periods as it performs write-heavy operations
• The migration preserves all historical data including dates, status, and user relationships
• If migrating large datasets, you may need to call this function multiple times until migrated_count returns 0
• Ensure the Neo4j driver has sufficient permissions to create nodes and relationships

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function migrate_approval_data 98.9% similar

  Migrates legacy single-step approval records in Neo4j to a new multi-step approval model by creating ApprovalStep nodes and Approver nodes with proper relationships.

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

• function migrate_approval_cycles 88.7% similar

  Migrates legacy Approval nodes and their related structures to the new ApprovalCycle model in a Neo4j graph database, creating ApprovalCycle nodes and ApproverAssignment nodes while maintaining relationships with DocumentVersion nodes.

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

• function migrate_review_assignments 70.5% similar

  Migrates legacy review data from direct REVIEWED_BY relationships to a new ReviewerAssignment node model in a Neo4j graph database.

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

• function update_review_comment_relationships 64.0% similar

  Migrates Neo4j ReviewComment nodes to use explicit parent/child relationships (PARENT_COMMENT and HAS_REPLY) based on existing parent_comment_uid properties.

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

• function migrate_audit_trail_v1 61.4% similar

  Migrates existing audit events from a legacy format to a new AuditTrail structure in a Neo4j database, returning migration statistics.

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