数据库迁移中的数据一致性保障与校验机制

import uuid

class IdempotentMigration:
    def __init__(self, db):
        self.db = db
    
    def generate_request_id(self):
        return str(uuid.uuid4())
    
    def check_request_id(self, request_id):
        result = self.db.execute(
            "SELECT COUNT(*) FROM migration_requests WHERE request_id = %s",
            (request_id,)
        )
        return result.fetchone()[0] > 0
    
    def save_request_id(self, request_id):
        self.db.execute(
            "INSERT INTO migration_requests (request_id, status, created_at) VALUES (%s, %s, NOW())",
            (request_id, 'processing')
        )
    
    def update_request_status(self, request_id, status):
        self.db.execute(
            "UPDATE migration_requests SET status = %s, updated_at = NOW() WHERE request_id = %s",
            (status, request_id)
        )
    
    def migrate_with_idempotency(self, data, request_id=None):
        if request_id is None:
            request_id = self.generate_request_id()
        
        if self.check_request_id(request_id):
            print(f"Request {request_id} already processed")
            return {'success': True, 'message': 'Already processed'}
        
        try:
            self.save_request_id(request_id)
            
            with self.db.begin():
                self._migrate_data(data)
            
            self.update_request_status(request_id, 'success')
            return {'success': True, 'message': 'Migration successful'}
        
        except Exception as e:
            self.update_request_status(request_id, 'failed')
            return {'success': False, 'message': str(e)}
    
    def _migrate_data(self, data):
        for record in data:
            existing = self.db.execute(
                "SELECT id FROM target_table WHERE id = %s",
                (record['id'],)
            ).fetchone()
            
            if existing:
                self.db.execute(
                    "UPDATE target_table SET name = %s, email = %s WHERE id = %s",
                    (record['name'], record['email'], record['id'])
                )
            else:
                self.db.execute(
                    "INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s)",
                    (record['id'], record['name'], record['email'])
                )

import redis
import time

class DistributedLock:
    def __init__(self, redis_client, lock_key, expire_time=30):
        self.redis = redis_client
        self.lock_key = lock_key
        self.expire_time = expire_time
        self.lock_value = None
    
    def acquire(self, timeout=10):
        end_time = time.time() + timeout
        
        while time.time() < end_time:
            self.lock_value = str(uuid.uuid4())
            result = self.redis.set(
                self.lock_key,
                self.lock_value,
                nx=True,
                ex=self.expire_time
            )
            
            if result:
                return True
            
            time.sleep(0.1)
        
        return False
    
    def release(self):
        if self.lock_value:
            script = """
                if redis.call("get", KEYS[1]) == ARGV[1] then
                    return redis.call("del", KEYS[1])
                else
                    return 0
                end
            """
            self.redis.eval(script, 1, self.lock_key, self.lock_value)
    
    def __enter__(self):
        self.acquire()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

class CheckpointMigration:
    def __init__(self, db):
        self.db = db
    
    def get_checkpoint(self, migration_id):
        result = self.db.execute(
            "SELECT position, status FROM migration_checkpoints WHERE migration_id = %s",
            (migration_id,)
        ).fetchone()
        
        if result:
            return {'position': result[0], 'status': result[1]}
        return {'position': 0, 'status': 'pending'}
    
    def set_checkpoint(self, migration_id, position, status):
        existing = self.db.execute(
            "SELECT id FROM migration_checkpoints WHERE migration_id = %s",
            (migration_id,)
        ).fetchone()
        
        if existing:
            self.db.execute(
                "UPDATE migration_checkpoints SET position = %s, status = %s, updated_at = NOW() WHERE migration_id = %s",
                (position, status, migration_id)
            )
        else:
            self.db.execute(
                "INSERT INTO migration_checkpoints (migration_id, position, status, created_at) VALUES (%s, %s, %s, NOW())",
                (migration_id, position, status)
            )
    
    def run_migration(self, migration_id, data, batch_size=1000):
        checkpoint = self.get_checkpoint(migration_id)
        start_pos = checkpoint['position']
        total_records = len(data)
        
        print(f"Resuming migration from position {start_pos}")
        
        for i in range(start_pos, total_records, batch_size):
            batch = data[i:i+batch_size]
            
            try:
                with self.db.begin():
                    self._process_batch(batch)
                
                self.set_checkpoint(migration_id, i + len(batch), 'processing')
                print(f"Migrated {i + len(batch)}/{total_records} records")
                
            except Exception as e:
                print(f"Error at position {i}: {e}")
                self.set_checkpoint(migration_id, i, 'failed')
                raise
        
        self.set_checkpoint(migration_id, total_records, 'completed')
        print("Migration completed successfully")
    
    def _process_batch(self, batch):
        for record in batch:
            self.db.execute(
                "INSERT INTO target_table (id, name, email) VALUES (%s, %s, %s) ON DUPLICATE KEY UPDATE name = %s, email = %s",
                (record['id'], record['name'], record['email'], record['name'], record['email'])
            )

class CountValidator:
    def __init__(self, source_db, target_db):
        self.source_db = source_db
        self.target_db = target_db
    
    def get_source_count(self, table_name):
        result = self.source_db.execute(f"SELECT COUNT(*) FROM {table_name}")
        return result.fetchone()[0]
    
    def get_target_count(self, table_name):
        result = self.target_db.execute(f"SELECT COUNT(*) FROM {table_name}")
        return result.fetchone()[0]
    
    def validate(self, table_name):
        source_count = self.get_source_count(table_name)
        target_count = self.get_target_count(table_name)
        
        is_valid = source_count == target_count
        
        return {
            'table': table_name,
            'source_count': source_count,
            'target_count': target_count,
            'is_valid': is_valid,
            'difference': source_count - target_count
        }

import hashlib

class ContentValidator:
    def __init__(self, source_db, target_db):
        self.source_db = source_db
        self.target_db = target_db
    
    def generate_record_hash(self, record):
        sorted_keys = sorted(record.keys())
        hash_string = '|'.join(f"{k}={record[k]}" for k in sorted_keys)
        return hashlib.md5(hash_string.encode()).hexdigest()
    
    def get_sample_data(self, db, table_name, sample_size=100):
        result = db.execute(
            f"SELECT * FROM {table_name} ORDER BY id LIMIT %s",
            (sample_size,)
        )
        columns = [desc[0] for desc in result.description]
        return [dict(zip(columns, row)) for row in result.fetchall()]
    
    def validate_sample(self, table_name, sample_size=100):
        source_sample = self.get_sample_data(self.source_db, table_name, sample_size)
        target_sample = self.get_sample_data(self.target_db, table_name, sample_size)
        
        mismatches = []
        
        for source_record, target_record in zip(source_sample, target_sample):
            source_hash = self.generate_record_hash(source_record)
            target_hash = self.generate_record_hash(target_record)
            
            if source_hash != target_hash:
                mismatches.append({
                    'id': source_record.get('id'),
                    'source': source_record,
                    'target': target_record
                })
        
        return {
            'table': table_name,
            'sample_size': sample_size,
            'mismatches': mismatches,
            'is_valid': len(mismatches) == 0
        }
    
    def validate_full(self, table_name):
        source_count = self.source_db.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
        target_count = self.target_db.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()[0]
        
        if source_count != target_count:
            return {
                'table': table_name,
                'is_valid': False,
                'message': f"Count mismatch: source={source_count}, target={target_count}"
            }
        
        batch_size = 1000
        mismatches = []
        
        for offset in range(0, source_count, batch_size):
            source_batch = self.source_db.execute(
                f"SELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s",
                (batch_size, offset)
            ).fetchall()
            
            target_batch = self.target_db.execute(
                f"SELECT * FROM {table_name} ORDER BY id LIMIT %s OFFSET %s",
                (batch_size, offset)
            ).fetchall()
            
            source_columns = [desc[0] for desc in source_batch.description]
            target_columns = [desc[0] for desc in target_batch.description]
            
            for source_row, target_row in zip(source_batch.fetchall(), target_batch.fetchall()):
                source_record = dict(zip(source_columns, source_row))
                target_record = dict(zip(target_columns, target_row))
                
                source_hash = self.generate_record_hash(source_record)
                target_hash = self.generate_record_hash(target_record)
                
                if source_hash != target_hash:
                    mismatches.append({
                        'id': source_record.get('id'),
                        'source': source_record,
                        'target': target_record
                    })
        
        return {
            'table': table_name,
            'total_records': source_count,
            'mismatches': mismatches,
            'is_valid': len(mismatches) == 0
        }

class StructureValidator:
    def __init__(self, source_db, target_db):
        self.source_db = source_db
        self.target_db = target_db
    
    def get_table_structure(self, db, table_name):
        result = db.execute(f"DESCRIBE {table_name}")
        columns = []
        for row in result.fetchall():
            columns.append({
                'name': row[0],
                'type': row[1],
                'null': row[2],
                'key': row[3],
                'default': row[4],
                'extra': row[5]
            })
        return columns
    
    def validate(self, table_name):
        source_structure = self.get_table_structure(self.source_db, table_name)
        target_structure = self.get_table_structure(self.target_db, table_name)
        
        source_columns = {col['name']: col for col in source_structure}
        target_columns = {col['name']: col for col in target_structure}
        
        missing_columns = [name for name in source_columns if name not in target_columns]
        extra_columns = [name for name in target_columns if name not in source_columns]
        type_mismatches = []
        
        for name in source_columns:
            if name in target_columns:
                source_type = source_columns[name]['type']
                target_type = target_columns[name]['type']
                if source_type != target_type:
                    type_mismatches.append({
                        'column': name,
                        'source_type': source_type,
                        'target_type': target_type
                    })
        
        return {
            'table': table_name,
            'missing_columns': missing_columns,
            'extra_columns': extra_columns,
            'type_mismatches': type_mismatches,
            'is_valid': not missing_columns and not extra_columns and not type_mismatches
        }

{
    "migration_id": "migration_001",
    "validation_time": "2024-01-15T10:00:00Z",
    "tables": [
        {
            "name": "users",
            "count_validation": {
                "source_count": 10000,
                "target_count": 10000,
                "is_valid": true
            },
            "content_validation": {
                "sample_size": 100,
                "mismatches": 0,
                "is_valid": true
            },
            "structure_validation": {
                "missing_columns": [],
                "extra_columns": [],
                "type_mismatches": [],
                "is_valid": true
            },
            "overall_status": "PASS"
        }
    ],
    "overall_status": "PASS",
    "total_tables": 1,
    "passed_tables": 1,
    "failed_tables": 0
}