🔍 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 (which directly linked to users) into a new hierarchical structure with ApprovalStep and Approver nodes. It checks for approvals without steps, creates default steps with sequence order 1, migrates approver relationships, and preserves historical data like dates and statuses. The migration is batched (1000 records at a time) and protected by a cooldown decorator to prevent excessive execution.

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: randomUUID(),
                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: randomUUID(),
                approver_uid: u.UID,
                approver_name: u.Name,
                status: 'COMPLETED',
                role: 'Approver',
                assigned_date: 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}")
        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 an active, authenticated connection to a Neo4j database containing Approval, User, ApprovalStep, and Approver node types.

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 nodes successfully migrated in this execution). On success with no records to migrate, migrated_count is 0. On error, success is False and message contains the exception details.

Dependencies

• neo4j
• CDocs

Required Imports

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

Usage Example

from neo4j import GraphDatabase
from typing import Dict, Any
from CDocs import guard_execution

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

try:
    # 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']}")
finally:
    driver.close()

Best Practices

• The function is protected by a guard_execution decorator with 5000ms cooldown to prevent rapid repeated executions
• Migration is batched with LIMIT 1000 to prevent memory issues with large datasets; call multiple times if needed
• Always check the 'success' field in the return dictionary before assuming migration completed
• The function is idempotent - it only migrates approvals that don't already have steps, safe to run multiple times
• Ensure proper error logging is configured as the function uses a module-level logger for exception tracking
• Close the Neo4j driver connection after migration is complete to free resources
• Consider running this migration during low-traffic periods as it creates multiple nodes and relationships
• The migration preserves original dates (startDate, completionDate) and status information from old approvals
• All new nodes receive unique UIDs via randomUUID() function in Neo4j
• The migration assumes a 1:1 mapping of old approvals to new single-step approvals with all_must_approve=true

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function migrate_approval_cycles 89.9% 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 get_approval_statistics_v1 60.7% similar

  Retrieves comprehensive approval statistics from a Neo4j graph database, including counts of total, pending, completed, approved, and rejected approval cycles and decisions, along with calculated approval rates.

  From: /tf/active/vicechatdev/CDocs/controllers/approval_controller.py

• function initialize_approval_workflow_types 59.7% similar

  Initializes default approval workflow types in a Neo4j database by creating or updating four predefined workflow type nodes (STANDARD, DEPARTMENT, QUALITY, REGULATORY) with their configurations.

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

• function get_approval_statistics 58.0% similar

  Retrieves and calculates approval cycle statistics for a specified date range, including counts by status, average completion time, and overdue approvals.

  From: /tf/active/vicechatdev/CDocs/controllers/approval_controller_bis.py

• function migrate_audit_events 57.9% similar

  Migrates orphaned AuditEvent nodes from CDocs nodes to an AuditTrail node in a Neo4j graph database by deleting old relationships and creating new ones.

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