Dataverse-Python-Real-World-Usecases

Legacy System → Staging Database → Dataverse
    (Extract)    (Transform)        (Load)

import pandas as pd
import time
from PowerPlatform.Dataverse.client import DataverseClient
from PowerPlatform.Dataverse.core.errors import DataverseError
from azure.identity import DefaultAzureCredential

class DataMigrationPipeline:
    """Migrate data from legacy system to Dataverse."""
    
    def __init__(self, org_url: str):
        self.client = DataverseClient(
            base_url=org_url,
            credential=DefaultAzureCredential()
        )
        self.success_records = []
        self.failed_records = []
    
    def extract_from_legacy(self, legacy_db_connection, query: str):
        """Extract data from source system."""
        return pd.read_sql(query, legacy_db_connection)
    
    def transform_accounts(self, df: pd.DataFrame) -> list:
        """Transform source data to Dataverse schema."""
        payloads = []
        
        for _, row in df.iterrows():
            # Map source fields to Dataverse
            payload = {
                "name": row["company_name"][:100],  # Limit to 100 chars
                "telephone1": row["phone"],
                "websiteurl": row["website"],
                "revenue": float(row["annual_revenue"]) if row["annual_revenue"] else None,
                "numberofemployees": int(row["employees"]) if row["employees"] else None,
                # Track source ID for reconciliation
                "new_sourcecompanyid": str(row["legacy_id"]),
                "new_importsequencenumber": row["legacy_id"]
            }
            payloads.append(payload)
        
        return payloads
    
    def load_to_dataverse(self, payloads: list, batch_size: int = 200):
        """Load data to Dataverse with error tracking."""
        total = len(payloads)
        
        for i in range(0, total, batch_size):
            batch = payloads[i:i + batch_size]
            
            try:
                ids = self.client.create("account", batch)
                self.success_records.extend(ids)
                print(f"✓ Created {len(ids)} records ({len(self.success_records)}/{total})")
                
                # Prevent rate limiting
                time.sleep(0.5)
                
            except DataverseError as e:
                self.failed_records.extend(batch)
                print(f"✗ Batch failed: {e.message}")
    
    def reconcile_migration(self, df: pd.DataFrame):
        """Verify migration and track results."""
        
        # Query created records
        created_accounts = self.client.get(
            "account",
            filter="new_importsequencenumber ne null",
            select=["accountid", "new_sourcecompanyid", "new_importsequencenumber"],
            top=10000
        )
        
        created_df = pd.DataFrame(list(created_accounts))
        
        # Update source table with Dataverse IDs
        merged = df.merge(
            created_df,
            left_on="legacy_id",
            right_on="new_importsequencenumber"
        )
        
        print(f"Successfully migrated {len(merged)} accounts")
        print(f"Failed: {len(self.failed_records)} records")
        
        return {
            "total_source": len(df),
            "migrated": len(merged),
            "failed": len(self.failed_records),
            "success_rate": len(merged) / len(df) * 100
        }

# Usage
pipeline = DataMigrationPipeline("https://myorg.crm.dynamics.com")

# Extract
source_data = pipeline.extract_from_legacy(
    legacy_connection,
    "SELECT id, company_name, phone, website, annual_revenue, employees FROM companies"
)

# Transform
payloads = pipeline.transform_accounts(source_data)

# Load
pipeline.load_to_datave