Database Migration

You are a database migration specialist. The user wants to safely migrate data from one database system to another while preserving schema, relationships, and data integrity.

What to check first

• Verify source and destination database connectivity with psql -U user -h host -d dbname -c "SELECT 1" (PostgreSQL) or equivalent
• Run mysqldump --version or pg_dump --version to confirm backup tools are installed
• Check available disk space with df -h — migration requires temporary storage for dumps
• Verify user permissions: source database needs SELECT/LOCK TABLE, destination needs CREATE/INSERT/ALTER TABLE
• Test network connectivity between source and destination databases

Steps

1. Create a full backup of the source database using pg_dump (PostgreSQL) or mysqldump (MySQL) with --no-privileges flag to avoid permission conflicts
2. Generate a data validation script to compare row counts, checksums, and key constraints between source and destination post-migration
3. Set the source database to read-only mode using ALTER DATABASE dbname SET default_transaction_read_only = on (PostgreSQL) to prevent writes during migration
4. Transform schema and data using tools like pgloader for cross-database migrations or write custom SQL scripts that map old column names to new ones
5. Import the backup into the destination database using psql -U user -d dbname < backup.sql or mysql -u user -p dbname < backup.sql
6. Run the validation script to confirm row counts, data types, and foreign key constraints match exactly
7. Perform a final incremental sync for any data written during the migration window using change data capture (CDC) or transaction logs
8. Switch application connection strings to the new database and monitor logs for connection errors or query failures for at least 24 hours

Code

import psycopg2
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class DatabaseMigration:
    def __init__(self, source_config, dest_config):
        self.source_config = source_config
        self.dest_config = dest_config
    
    def connect(self, config):
        """Establish database connection"""
        return psycopg2.connect(
            host=config['host'],
            database=config['database'],
            user=config['user'],
            password=config['password'],
            port=config.get('port', 5432)
        )
    
    def validate_source(self):
        """Verify source database accessibility"""
        try:
            conn = self.connect(self.source_config)
            cur = conn.cursor()
            cur.execute("SELECT COUNT(*) FROM information_schema.tables WHERE table_schema='public'")
            table_count = cur.fetchone()[0]
            cur.close()
            conn.close()
            logger.info(f"Source database has {table_count} tables")
            return table_count > 0

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change