Client Data Migration - Dal CRM Documentation

This guide provides a comprehensive approach to migrating large volumes of client data to DAL using our bulk import APIs. Whether you’re migrating from a legacy system or consolidating data from multiple sources, this guide will help you implement a robust migration strategy.

Overview

DAL’s bulk import API allows you to efficiently migrate large datasets while maintaining data integrity and system performance. This guide covers:

Batch Processing: How to handle importing large datasets efficiently.
Data Validation: Ensuring data quality before import.
Error Handling: Robust retry mechanisms and failure recovery.
Performance Optimization: Best practices for optimal throughput when integrating with Dal.

Prerequisites

Before starting your migration, ensure you have:

API Credentials: Contact your DAL administrator for API key and tenant information.
Data Access: Access to your source data (CSV, database, etc.).
Network Connectivity: Stable connection to DAL’s API endpoints.

What We Expect From You

The migration process involves making HTTP calls to DAL’s bulk client endpoint. Your client data should be prepared and sent in batches to this endpoint. Once received, DAL’s backend system will process these batches, validate the data, and reflect the imported clients in the web portal for your compliance team’s review and management. To facilitate this, your development team should implement a dedicated service layer or a migration script that extracts data from your existing database and transforms it into the required format before sending it to DAL via HTTP requests.

Migration Examples

Python Implementation

Required Python Packages

pip install requests csv itertools

Configuration

API Endpoints

# Production endpoint
API_ENDPOINT = "https://prod.api.getdal.sa/services/crm/clients/bulk"

# UAT endpoint (for testing)
API_ENDPOINT = "https://app.api.getdal.sa/services/crm/clients/bulk"

Authentication Headers

HEADERS = {
    "x-api-key": "your-api-key-here",  # Provided by DAL admin
    "x-tenant": "your-tenant-name"     # Your organization identifier
}

Batch Configuration

# Recommended batch sizes based on environment
BATCH_SIZE = 5000  # (Maximum Allowed Size)

# Retry configuration
MAX_RETRIES = 3
RETRY_DELAY = 3  # seconds between retries

Duplicate Modes

Choose how to handle existing clients during migration:

Skip: Skip existing clients (recommended for migrations).
Update: Update existing clients with new data.

DUPLICATE_MODE = "Skip"  # Recommended for migrations

Data Mapping

Client Fields

Fields:

nameEn - Name in English (string)
nameAr - Name in Arabic (string)
type - Entity type (Individual/Organization)
email - Email address
accountOpenDate - Account opening date (ISO format)
dobGregorian - Date of birth (ISO format)
dateOfIncorporationGregorian - Incorporation date (ISO format)
nationalityCode - ISO 3166-1 alpha-2 country code
status - Client status
identities - Array of identities with a minimum size of 1

Note: For the complete request payload structure and field specifications, refer to the Bulk Insertion API Endpoint documentation.

Nationality Mapping

Map your nationality values to ISO 3166-1 alpha-2 codes:

NATIONALITY_MAP = {
    "سعودي": "SA",      # Saudi Arabia
    "باكستاني": "PK",   # Pakistan
    "أردني": "JO",      # Jordan
    "مصري": "EG",       # Egypt
    "أمريكي": "US",     # United States
    "بريطاني": "GB",    # United Kingdom
    # Add all nationality mappings as needed per ISO 3166-1 alpha-2 codes
}

Data Validation

Date Format Conversion

Convert dates to ISO format (YYYY-MM-DD):

def to_iso_date(date_str):
    # Convert date string to ISO format
    if not date_str or not date_str.strip():
        return None
    try:
        # Handle common date formats
        formats = ["%m/%d/%Y", "%d/%m/%Y", "%Y-%m-%d", "%d-%m-%Y"]
        for fmt in formats:
            try:
                return datetime.strptime(date_str.strip(), fmt).strftime("%Y-%m-%d")
            except ValueError:
                continue
        return None
    except ValueError:
        return None

Client Validation

REQUIRED_STR_FIELDS = ["type", "identities", "nameEn", "nameAr"]

def is_valid_client(client):
    # Validate client data before sending
    for field in REQUIRED_STR_FIELDS:
        if not client.get(field) or not isinstance(client.get(field), str) or not client.get(field).strip():
            return False
    return True

Batch Processing

Reading Data in Batches

def read_in_batches(file_path, batch_size):
    # Reading from CSV file in batches, real scenario should include fetching data from database
    with open(file_path, newline='', encoding="utf-8") as f:
        reader = csv.DictReader(f)
        while True:
            batch = list(islice(reader, batch_size))
            if not batch:
                break
            yield batch

Data Mapping Function

def map_to_client(row):
    # Map database row to DAL client format
    client = {
        "type": row.get("entity type"),
        "nameEn": row.get("name"),
        "nameAr": row.get("name ar"),
        "identities": [
            {
                "type": row.get("Id type"),
                "number": row.get("Id number"),
            }
        ],
        "email": None,
        "idExpirationDate": to_iso_date(row.get("ID expiration")),
        "accountOpenDate": to_iso_date(row.get("Account open date")),
        "dobGregorian": to_iso_date(row.get("Date of birth")),
        "dateOfIncorporationGregorian": to_iso_date(row.get("Date of incorporation")),
        "nationalityCode": NATIONALITY_MAP.get(row.get("Nationality", "").strip()) if row.get("Nationality") else None,
        "status": row.get("Client status"),
    }

    return client

Error Handling and Retries

Robust Batch Sending

def send_batch(clients, duplicate_mode="Skip"):
    # Send batch with retry mechanism
    payload = {
        "clients": clients,
        "duplicateMode": duplicate_mode,
    }

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = requests.post(API_ENDPOINT, headers=HEADERS, json=payload, timeout=60)
            if response.status_code == 201:
                return True
            else:
                print(f"Failed (status {response.status_code}): {response.text}")
        except requests.exceptions.Timeout:
            print(f"Timeout on attempt {attempt}")
        except requests.exceptions.ConnectionError:
            print(f"Connection error on attempt {attempt}")
        except Exception as e:
            print(f"Error sending batch: {e}")

        if attempt < MAX_RETRIES:
            sleep(RETRY_DELAY)

    print("Giving up on this batch after max retries.")
    return False

Complete Migration Script (Python Example, Typescript Example Is Provided In The Next Section)

import csv
import json
import requests
from itertools import islice
from time import sleep, time
from datetime import datetime

# Configuration
CSV_FILE = "datasets/clients.csv"
BATCH_SIZE = 5000  # Adjust based on your environment; our servers can handle receiving up to 5000 per request
API_ENDPOINT = "https://<environment>.api.getdal.sa/services/crm/clients/bulk"

HEADERS = {
    "x-api-key": "your-api-key-here",
    "x-tenant": "your-tenant-name"
}

DUPLICATE_MODE = "Skip"
MAX_RETRIES = 3
RETRY_DELAY = 3

NATIONALITY_MAP = {
    "سعودي": "SA",
    "باكستاني": "PK",
    "أردني": "JO",
    "مصري": "EG",
}

REQUIRED_STR_FIELDS = ["type", "identities", "nameEn", "nameAr"]

def to_iso_date(date_str):
    if not date_str or not date_str.strip():
        return None
    try:
        return datetime.strptime(date_str.strip(), "%m/%d/%Y").strftime("%Y-%m-%d")
    except ValueError:
        return None

def read_in_batches(file_path, batch_size):
    with open(file_path, newline='', encoding="utf-8") as f:
        reader = csv.DictReader(f)
        while True:
            batch = list(islice(reader, batch_size))
            if not batch:
                break
            yield batch

def map_to_client(row):
    client = {
        "type": row.get("entity type"),
        "nameEn": row.get("name"),
        "nameAr": row.get("name ar"),
        "identities": [
            {
                "type": row.get("Id type"),
                "number": row.get("Id number"),
            }
        ],
        "email": None,
        "idExpirationDate": to_iso_date(row.get("ID expiration")),
        "accountOpenDate": to_iso_date(row.get("Account open date")),
        "dobGregorian": to_iso_date(row.get("Date of birth")),
        "dateOfIncorporationGregorian": to_iso_date(row.get("Date of incorporation")),
        "nationalityCode": NATIONALITY_MAP.get(row.get("Nationality", "").strip()) if row.get("Nationality") else None,
        "status": row.get("Client status"),
    }

    return client

def is_valid_client(client):
    for field in REQUIRED_STR_FIELDS:
        if not client.get(field) or not isinstance(client.get(field), str) or not client.get(field).strip():
            return False
    return True

def send_batch(clients):
    payload = {
        "clients": clients,
        "duplicateMode": DUPLICATE_MODE,
    }
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = requests.post(API_ENDPOINT, headers=HEADERS, json=payload, timeout=60)
            if response.status_code == 201:
                return True
            else:
                print(f"Failed (status {response.status_code}): {response.text}")
        except Exception as e:
            print(f"Error sending batch: {e}")

        sleep(RETRY_DELAY)

    print("Giving up on this batch after max retries.")
    return False

def main():
    total_sent = 0
    total_skipped = 0
    batch_number = 0
    start_time = time()

    print(f"Migration started at {datetime.fromtimestamp(start_time)}")

    for batch_rows in read_in_batches(CSV_FILE, BATCH_SIZE):
        batch_number += 1
        batch_start = time()

        clients_mapped = [map_to_client(row) for row in batch_rows]
        valid_clients = [c for c in clients_mapped if is_valid_client(c)]
        skipped_count = len(clients_mapped) - len(valid_clients)
        total_skipped += skipped_count

        if skipped_count > 0:
            print(f"Batch {batch_number}: Skipped {skipped_count} invalid rows")

        if not valid_clients:
            continue

        success = send_batch(valid_clients)
        if success:
            total_sent += len(valid_clients)

        batch_end = time()
        print(f"Batch {batch_number} processed in {batch_end - batch_start:.2f}s")

    end_time = time()
    print(f"Migration completed!")
    print(f"Total clients sent: {total_sent}")
    print(f"Total skipped: {total_skipped}")
    print(f"Total elapsed time: {end_time - start_time:.2f}s")
    print(f"Migration ended at {datetime.fromtimestamp(end_time)}")

if __name__ == "__main__":
    main()

Best Practices When Implementing Your Own Migrator Service/Script

1. Error Handling

Implement comprehensive retry logic
Log all failures for debugging
Validate data before sending
Handle network timeouts gracefully
Keep track of batches to retry failed ones later

2. Performance Monitoring

Track processing time per batch
Log progress regularly
Set up alerts/logs for failures

3. Data Quality

Validate required fields before sending
Clean and normalize data
Handle missing or invalid dates
Map nationalities correctly

Troubleshooting

Common Issues

Authentication Errors: Verify API key and tenant name
Validation Failures: Check required fields and data formats
Timeout Errors: Reduce batch size or increase timeout
Memory Issues: Process data in smaller batches

Debug Mode

Enable detailed logging for troubleshooting:

import logging

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

def send_batch(clients):
    logger.debug(f"Sending batch with {len(clients)} clients")
    # ... rest of implementation

Post-Migration

Case Generation

After completing client migration, log into the portal to review historical data case matching results.

Verification

Verify your migration by:

Checking client counts in DAL dashboard
Validating sample records
Testing case generation
Reviewing audit logs

Support

For migration assistance:

Review this guide thoroughly
Test with small datasets first and keep track of your batch successes and failures
Contact your DAL administrator or development team for support via the dedicated Slack channel

API Reference

Bulk Import Endpoint

POST /services/crm/clients/bulk Import multiple clients in a single request. Headers:

x-api-key: Your API key
x-tenant: Your tenant name
Content-Type: application/json

Request Body:

{
  "clients": [
    {
      "identities": [{ "type": "NationalIdentity", "number": "1234567890" }],
      "nameEn": "John Michael Doe",
      "nameAr": "جون مايكل دو",
      "nationalityCode": "SA",
      "dobGregorian": "1990-01-01"
    }
  ],
  "duplicateMode": "Skip"
}

Response:

{
  "success": true,
  "message": "Clients imported successfully"
}

TypeScript Implementation

For Node.js environments, here’s a TypeScript implementation:

Required Packages

npm install axios csv-parser fs-extra
npm install -D @types/node typescript

Migration Script (TypeScript Example)

import axios from "axios";
import csv from "csv-parser";
import fs from "fs-extra";
import { createReadStream } from "fs";

interface Client {
  idNumber: string;
  nameEn: string;
  nameAr: string;
  nationalityCode?: string;
  dobGregorian?: string;
  // ... other optional fields
}

const config = {
  apiEndpoint: "https://<environment>.api.getdal.sa/services/crm/clients/bulk",
  apiKey: "your-api-key-here",
  tenant: "your-tenant-name",
  batchSize: 5000,
  maxRetries: 3,
  retryDelay: 3000,
};

const nationalityMap: Record<string, string> = {
  سعودي: "SA",
  باكستاني: "PK",
  أردني: "JO",
  مصري: "EG",
};

// Note: In a real-world scenario, client data would typically be fetched from a database rather than a CSV file

async function processCsvFile(filePath: string): Promise<void> {
  const clients: Client[] = [];

  return new Promise((resolve, reject) => {
    createReadStream(filePath)
      .pipe(csv())
      .on("data", (row) => {
        const client = mapToClient(row);
        if (isValidClient(client)) {
          clients.push(client);
        }
      })
      .on("end", () => {
        processBatches(clients).then(resolve).catch(reject);
      })
      .on("error", reject);
  });
}

function mapToClient(row: any): Client {
  const client: Client = {
    type: row["entity type"],
    nameEn: row["name"],
    nameAr: row["name ar"],
    identities: [
      {
        type: row["Id type"],
        number: row["Id number"],
      },
    ],
    nationalityCode: nationalityMap[row["Nationality"]?.trim()],
    dobGregorian: toIsoDate(row["Date of birth"]),
  };

  return client;
}

function toIsoDate(dateStr: string): string | undefined {
  if (!dateStr?.trim()) return undefined;
  try {
    return new Date(dateStr.trim()).toISOString().split("T")[0];
  } catch {
    return undefined;
  }
}

function isValidClient(client: Client): boolean {
  const required = ["type", "identities", "nameEn", "nameAr"];
  return required.every((field) =>
    client[field as keyof Client]?.toString().trim(),
  );
}

async function sendBatch(clients: Client[]): Promise<boolean> {
  const payload = {
    clients,
    duplicateMode: "Skip" as const,
  };

  for (let attempt = 1; attempt <= config.maxRetries; attempt++) {
    try {
      const response = await axios.post(config.apiEndpoint, payload, {
        headers: {
          "x-api-key": config.apiKey,
          "x-tenant": config.tenant,
          "Content-Type": "application/json",
        },
        timeout: 60000,
      });

      if (response.status === 201) {
        return true;
      }
    } catch (error: any) {
      console.error(`Attempt ${attempt} failed:`, error.message);
      if (attempt < config.maxRetries) {
        await new Promise((resolve) => setTimeout(resolve, config.retryDelay));
      }
    }
  }
  return false;
}

async function processBatches(clients: Client[]): Promise<void> {
  let totalSent = 0;
  let totalSkipped = 0;
  let batchNumber = 0;

  for (let i = 0; i < clients.length; i += config.batchSize) {
    batchNumber++;
    const batch = clients.slice(i, i + config.batchSize);

    const validClients = batch.filter(isValidClient);
    const skippedCount = batch.length - validClients.length;
    totalSkipped += skippedCount;

    if (skippedCount > 0) {
      console.log(`Batch ${batchNumber}: Skipped ${skippedCount} invalid rows`);
    }

    if (validClients.length === 0) {
      continue;
    }

    const success = await sendBatch(validClients);
    if (success) {
      totalSent += validClients.length;
    }

    console.log(
      `Batch ${batchNumber} processed (${validClients.length} clients)`,
    );
  }

  console.log(`Migration completed!`);
  console.log(`Total clients sent: ${totalSent}`);
  console.log(`Total skipped: ${totalSkipped}`);
}

// Usage
async function main() {
  try {
    await processCsvFile("datasets/clients.csv");
  } catch (error) {
    console.error("Migration failed:", error);
  }
}

main();

Key Differences from Python

Type Safety: Full TypeScript types for better development experience.
Async/Await: Native promise handling.
Streaming: Efficient data processing with streams.
Error Handling: Structured error handling with try/catch.

Requesting Help In Implementing The Scripts

You can request help by contacting your DAL administrator or support team for migration assistance, please include details about your tech stack, areas of difficulty, and specific ways our development team can help.

​Overview

​Prerequisites

​What We Expect From You

​Migration Examples

​Python Implementation

​Required Python Packages

​Configuration

API Endpoints

Authentication Headers

Batch Configuration

Duplicate Modes

​Data Mapping

​Client Fields

​Nationality Mapping

​Data Validation

​Date Format Conversion

​Client Validation

​Batch Processing

​Reading Data in Batches

​Data Mapping Function

​Error Handling and Retries

​Robust Batch Sending

​Complete Migration Script (Python Example, Typescript Example Is Provided In The Next Section)

​Best Practices When Implementing Your Own Migrator Service/Script

​1. Error Handling

​2. Performance Monitoring

​3. Data Quality

​Troubleshooting

​Common Issues

​Debug Mode

​Post-Migration

​Case Generation

​Verification

​Support

​API Reference

​Bulk Import Endpoint

​TypeScript Implementation

​Required Packages

​Migration Script (TypeScript Example)

​Key Differences from Python

​Requesting Help In Implementing The Scripts