Zero Downtime Migration with Dual Write Pattern in Python

from datetime import datetime
import json


class UserService:
    def __init__(self):
        self.legacy_db = {}
        self.new_db = {}
        self.migration_log = []

    def write_user(self, user_id, name, email):
        # Write to new system first
        user_record = {
            "id": user_id,
            "name": name,
            "email": email,
            "updated_at": datetime.utcnow().isoformat()
        }
        self.new_db[user_id] = user_record

        # Dual-write to legacy system
        self.legacy_db[user_id] = user_record.copy()
        self.migration_log.append(f"Wrote user {user_id} to both systems at {user_record['updated_at']}")

    def read_from_legacy(self, user_id):
        return self.legacy_db.get(user_id)

    def read_from_new(self, user_id):
        return self.new_db.get(user_id)

    def verify_consistency(self):
        mismatches = []
        for user_id in set(self.legacy_db) | set(self.new_db):
            if self.legacy_db.get(user_id) != self.new_db.get(user_id):
                mismatches.append(user_id)
        return len(mismatches) == 0, mismatches


if __name__ == "__main__":
    service = UserService()

    # Simulate migration traffic
    service.write_user("user-001", "Alice", "alice@example.com")
    service.write_user("user-002", "Bob", "bob@example.com")
    service.write_user("user-003", "Carol", "carol@example.com")

    # Simulate reads during migration
    print("Legacy read:", service.read_from_legacy("user-002"))
    print("New read:", service.read_from_new("user-002"))

    # Verify data consistency between both stores
    is_consistent, conflicts = service.verify_consistency()
    print(f"\nMigration log entries: {len(service.migration_log)}")
    print(f"All systems consistent: {is_consistent}")
    print(f"Conflict IDs: {conflicts}")