🔍 Code Extractor

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.

Purpose

This function performs a database schema migration for approval workflows in a document management system. It identifies old Approval structures that haven't been migrated, creates new ApprovalCycle nodes with standardized properties, and transforms old ApprovalStep/Approver relationships into ApproverAssignment nodes. The migration is batched (500 records at a time) to prevent memory issues and includes a cooldown mechanism to prevent excessive execution. It's designed to be idempotent, only migrating approvals that haven't already been converted.

Source Code

def migrate_approval_cycles(driver: Driver) -> Dict[str, Any]:
    """
    Migrate old approval structure to new ApprovalCycle model.
    
    Args:
        driver: Neo4j driver instance
        
    Returns:
        Dict containing migration results
    """
    try:
        with driver.session() as session:
            # Check for Approvals that need to be migrated to ApprovalCycles
            result = session.run(
                """
                MATCH (a:Approval)
                WHERE NOT EXISTS((a)-[:FOR_APPROVAL]->(:ApprovalCycle))
                RETURN count(a) as count
                """
            )
            record = result.single()
            
            if not record or record["count"] == 0:
                return {
                    "success": True,
                    "message": "No approval cycles need migration",
                    "migrated_count": 0
                }
            
            # Perform migration from old Approval to new ApprovalCycle
            migration_query = """
            MATCH (a:Approval)-[:FOR_APPROVAL]->(dv:DocumentVersion)
            WHERE NOT EXISTS((dv)-[:FOR_APPROVAL]->(:ApprovalCycle))
            WITH a, dv LIMIT 500
            
            // Create a new ApprovalCycle
            CREATE (ac:ApprovalCycle {
                UID: randomUUID(),
                status: a.status,
                createdAt: a.startDate,
                completionDate: a.completionDate,
                dueDate: a.dueDate,
                instructions: a.instructions,
                approvalType: 'STANDARD',
                sequential: false,
                requiredApprovalPercentage: 100,
                initiatedByUID: a.initiatedBy,
                properties: {}
            })
            
            // Connect document version to approval cycle
            CREATE (dv)-[:FOR_APPROVAL]->(ac)
            
            // Migrate approvers to approver assignments
            WITH a, ac
            OPTIONAL MATCH (a)-[:HAS_STEP]->(step:ApprovalStep)-[:HAS_APPROVER]->(apr:Approver)
            WHERE apr.approver_uid IS NOT NULL
            
            // Create approver assignments for each approver
            CREATE (aa:ApproverAssignment {
                UID: randomUUID(),
                approverUID: apr.approver_uid,
                name: apr.approver_name,
                role: apr.role,
                status: apr.status,
                decision: apr.decision,
                decisionDate: apr.decision_date,
                firstActivityDate: apr.assigned_date,
                instructions: '',
                sequence_order: 1
            })
            
            // Connect approver assignment to approval cycle
            CREATE (ac)-[:ASSIGNMENT]->(aa)
            
            RETURN count(a) as migrated
            """
            
            result = session.run(migration_query)
            record = result.single()
            
            return {
                "success": True,
                "message": f"Successfully migrated {record['migrated']} approval cycles",
                "migrated_count": record["migrated"]
            }
            
    except Exception as e:
        logger.error(f"Error migrating approval cycles: {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. This should be a properly configured neo4j.Driver object with valid credentials and connection to the target Neo4j database containing the approval structures to be migrated.

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 to ApprovalCycle nodes). On success with no migrations needed, migrated_count is 0. On error, success is False and message contains the exception details.

Dependencies

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

Required Imports

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

Conditional/Optional Imports

These imports are only needed under specific conditions:

from CDocs import guard_execution

from CDocs import db

import uuid

import traceback

Usage Example

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

# Configure logging
logger = logging.getLogger(__name__)

# Setup Neo4j connection
uri = "bolt://localhost:7687"
username = "neo4j"
password = "your_password"
driver = GraphDatabase.driver(uri, auth=(username, password))

try:
    # Execute migration
    result = migrate_approval_cycles(driver)
    
    if result["success"]:
        print(f"Migration successful: {result['message']}")
        print(f"Migrated {result['migrated_count']} approval cycles")
    else:
        print(f"Migration failed: {result['message']}")
finally:
    driver.close()

Best Practices

• Always close the Neo4j driver connection after migration to prevent resource leaks
• The function is decorated with guard_execution(cooldown_ms=5000), so avoid calling it more frequently than every 5 seconds
• Migration is batched at 500 records per execution - for large datasets, call the function multiple times until migrated_count returns 0
• Check the 'success' field in the return dictionary before assuming migration completed successfully
• Ensure proper error logging is configured as the function logs errors using the module-level logger
• The migration is idempotent - it only migrates Approval nodes that don't already have ApprovalCycle relationships, making it safe to run multiple times
• Test the migration on a backup or development database before running on production data
• Monitor the migrated_count to track progress across multiple executions for large datasets
• The function creates new UIDs using Neo4j's randomUUID() function, ensuring unique identifiers for new nodes

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function migrate_approval_data 89.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 get_approval_cycle 66.8% similar

  Retrieves detailed information about an approval cycle by its UID, with optional inclusion of comments and associated document details.

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

• function get_approval_statistics_v1 65.5% 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 create_approval_cycle 65.4% similar

  Creates a new approval cycle for a document, assigning approvers with configurable workflow options (sequential/parallel), instructions, and due dates.

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

• function get_document_approvals 63.2% similar

  Retrieves all approval cycles associated with a specific document, with optional filtering for active cycles only.

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