ClickUp API: Resume Failed Migration Without Duplicate Tasks in Python

ClickUp API: Resume a Failed Migration Without Duplicate Tasks (Python)

Category: API Architecture & DevOps • Reading Time: 12 mins • Updated for: ClickUp API v2 / v3

The Nightmare of Mid-Flight Migration Crashes

You fire up a Python script to migrate 5,000 legacy cards into ClickUp. At record 2,418, an unexpected network timeout, an unhandled HTTP 429 Rate Limit spike, or a malformed custom field crashes your process. If you blindly restart the script, ClickUp creates 2,417 duplicate tasks, wrecking sprint boards, polluting notifications, and forcing painful manual cleanup.

Building a reliable data pipeline demands idempotency: the mathematical guarantee that executing an operation multiple times produces the exact same outcome as running it once. If you already implemented a base migration as covered in our Trello to ClickUp Python Migration Script, this guide upgrades your setup into an enterprise-grade, fault-tolerant engine.

1. Core Architectural Pillars of Resilient Migrations

To survive network drops, API throttles, and process kills without creating ghost duplicates, your script requires four structural safeguards:

• Atomic Checkpoint Mapping (JSON/SQLite): Maintain an explicit key-value lookup (source_card_id → destination_task_id) flushed to disk immediately after each successful task creation.
• Source Key Injection (Metadata Recovery): Inject the source ID directly into a dedicated ClickUp Custom Field or append it as an HTML/Markdown comment in the task description. If your local mapping file is deleted, you can reconstruct it dynamically via the ClickUp search endpoint.
• Dynamic Exponential Backoff (HTTP 429): Automatically pause when throttled, read the Retry-After header, and inject jitter to prevent thundering herd collisions.
• Two-Pass Architecture with Quarantined Retries: Hard failures are isolated into a failed_records.json quarantine log instead of terminating the entire batch.

⚡ Interactive Migration Time & Rate Limit Estimator

Calculate estimated completion times and plan buffer intervals based on your ClickUp API plan limit.

Total Tasks

Rate Limit (req/min) Free / Standard Tier (100 RPM) Business / Enterprise (900 RPM)

Safety Throttle Delay (sec)

Estimated Run Duration: ~12.5 minutes | Safe Throughput: 240 tasks/min

2. Deep Dive: The Idempotent Execution Flow

An idempotent pipeline inspects its state before taking any external mutation action. Here is the operational loop implemented in the Python script:

1. Load Checkpoint: Read migration_checkpoint.json into an in-memory dict.
2. Evaluate Card: For each card in the source list:
   • If card['id'] in checkpoint: Skip creation, log an audit pass, increment skipped_count.
   • If not mapped: Send an HTTP POST payload to /api/v2/list/{list_id}/task.
3. Atomic Write: If creation succeeds (HTTP 200/201), store checkpoint[card_id] = response['id'] and immediately flush to the file system.
4. Fail-Soft Logging: If an unrecoverable HTTP status (e.g., 400 Bad Request, unmapped custom fields) occurs, dump the raw card payload into failed_cards.json and continue the execution thread. For missing custom field issues, review our deep dive on Mapping Trello Custom Fields to ClickUp Without Data Loss.

3. Production-Ready Python Script

The following script is fully self-contained. It incorporates a dry-run flag, HTTP 429 exponential backoff, persistent state mapping, and end-of-run reconciliation counters.

import os
import json
import time
import random
import logging
from typing import Dict, Any, List, Optional
import requests

# ---------------------------------------------------------
# Logging Configuration
# ---------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler("migration.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("ClickUpMigration")

# ---------------------------------------------------------
# Configuration Settings
# ---------------------------------------------------------
CLICKUP_API_TOKEN = os.getenv("CLICKUP_API_TOKEN", "pk_your_api_token_here")
CLICKUP_LIST_ID = os.getenv("CLICKUP_LIST_ID", "your_target_list_id")
CHECKPOINT_FILE = "migration_checkpoint.json"
FAILED_LOG_FILE = "failed_cards.json"
DRY_RUN = False  # Set to True to simulate without creating real tasks

BASE_URL = "https://api.clickup.com/api/v2"
HEADERS = {
    "Authorization": CLICKUP_API_TOKEN,
    "Content-Type": "application/json"
}

# ---------------------------------------------------------
# Checkpoint & State Management Helpers
# ---------------------------------------------------------
def load_checkpoint(filepath: str) -> Dict[str, str]:
    """Loads existing card_id -> task_id mappings from disk."""
    if os.path.exists(filepath):
        try:
            with open(filepath, "r", encoding="utf-8") as f:
                data = json.load(f)
                logger.info(f"Loaded {len(data)} existing checkpoints from {filepath}")
                return data
        except Exception as e:
            logger.error(f"Failed to read checkpoint file: {e}. Starting with an empty mapping.")
    return {}

def save_checkpoint_atomic(filepath: str, data: Dict[str, str]) -> None:
    """Writes mapping atomically using a temporary file to avoid corruption."""
    temp_file = f"{filepath}.tmp"
    with open(temp_file, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    os.replace(temp_file, filepath)

def log_failure(source_card: Dict[str, Any], reason: str) -> None:
    """Appends failed records to an isolated quarantine file."""
    failures = []
    if os.path.exists(FAILED_LOG_FILE):
        try:
            with open(FAILED_LOG_FILE, "r", encoding="utf-8") as f:
                failures = json.load(f)
        except Exception:
            failures = []
    
    failures.append({
        "card_id": source_card.get("id"),
        "card_name": source_card.get("name"),
        "error": reason,
        "timestamp": time.time()
    })
    
    with open(FAILED_LOG_FILE, "w", encoding="utf-8") as f:
        json.dump(failures, f, indent=2)

# ---------------------------------------------------------
# Resilient API Dispatcher with Backoff
# ---------------------------------------------------------
def create_clickup_task_with_retry(
    payload: Dict[str, Any], 
    max_retries: int = 5
) -> Optional[Dict[str, Any]]:
    """Dispatches task payload with exponential backoff for 429 rate limits."""
    if DRY_RUN:
        logger.info(f"[DRY-RUN] Simulating task creation: {payload.get('name')}")
        return {"id": f"dry_run_{int(time.time() * 1000)}"}

    url = f"{BASE_URL}/list/{CLICKUP_LIST_ID}/task"

    for attempt in range(1, max_retries + 1):
        try:
            response = requests.post(url, headers=HEADERS, json=payload, timeout=15)
            
            # Success
            if response.status_code in [200, 201]:
                return response.json()
            
            # Rate Limited (HTTP 429)
            elif response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                jitter = random.uniform(0.5, 1.5)
                sleep_duration = retry_after + jitter
                logger.warning(
                    f"429 Throttled. Backing off for {sleep_duration:.2f}s (Attempt {attempt}/{max_retries})"
                )
                time.sleep(sleep_duration)
            
            # Server-side transient errors
            elif response.status_code in [500, 502, 503, 504]:
                wait_time = (2 ** attempt) + random.uniform(0.1, 1.0)
                logger.warning(f"Server error {response.status_code}. Retrying in {wait_time:.2f}s...")
                time.sleep(wait_time)
            
            # Client-side fatal errors (400, 401, 403, 404)
            else:
                logger.error(f"Fatal API error {response.status_code}: {response.text}")
                return None

        except requests.exceptions.RequestException as exc:
            logger.warning(f"Network error on attempt {attempt}: {exc}")
            time.sleep((2 ** attempt) + 1)

    return None

# ---------------------------------------------------------
# Core Migration Runner
# ---------------------------------------------------------
def run_migration(source_cards: List[Dict[str, Any]]) -> None:
    """Executes the idempotent migration pipeline."""
    checkpoint_map = load_checkpoint(CHECKPOINT_FILE)
    
    total_cards = len(source_cards)
    created_count = 0
    skipped_count = 0
    failed_count = 0

    logger.info(f"Starting migration. Target records: {total_cards}")

    for idx, card in enumerate(source_cards, start=1):
        card_id = card.get("id")
        card_name = card.get("name", "Untitled Task")
        card_desc = card.get("desc", "")

        # 1. Idempotency Gate
        if card_id in checkpoint_map:
            logger.info(f"[{idx}/{total_cards}] SKIPPED: {card_id} already exists in ClickUp as {checkpoint_map[card_id]}")
            skipped_count += 1
            continue

        # 2. Metadata Redundancy: Embed source ID in description
        recovery_footer = f"\n\n---\n*Migrated from Trello Card ID: {card_id}*"
        complete_desc = f"{card_desc}{recovery_footer}"

        payload = {
            "name": card_name,
            "description": complete_desc,
            "status": "to do"
        }

        # 3. API Dispatch
        result = create_clickup_task_with_retry(payload)

        # 4. State Update
        if result and "id" in result:
            new_task_id = result["id"]
            checkpoint_map[card_id] = new_task_id
            save_checkpoint_atomic(CHECKPOINT_FILE, checkpoint_map)
            logger.info(f"[{idx}/{total_cards}] CREATED: Card {card_id} -> ClickUp {new_task_id}")
            created_count += 1
        else:
            log_failure(card, "API task creation failed after retries.")
            failed_count += 1

        # Gentle throttling to conserve rate limits
        time.sleep(0.15)

    # 5. Final Reconciliation Summary
    print("\n" + "="*50)
    print("MIGRATION RECONCILIATION AUDIT")
    print("="*50)
    print(f"Total Source Cards : {total_cards}")
    print(f"Skipped (Deduplicated) : {skipped_count}")
    print(f"Newly Created      : {created_count}")
    print(f"Failed Records     : {failed_count}")
    print("="*50)

    if failed_count > 0:
        print(f"⚠️ Inspect '{FAILED_LOG_FILE}' to isolate and retry failed records.")
        print(f"Detailed troubleshooting: https://saasethu.blogspot.com/2026/10/fix-trello-to-clickup-migration-failed-errors.html")

# ---------------------------------------------------------
# Execution Entry Point (Sample Data)
# ---------------------------------------------------------
if __name__ == "__main__":
    # Mock source payload simulating Trello API dump
    sample_source_cards = [
        {"id": "64a01f1", "name": "Design Checkout Flow", "desc": "Wireframe checkout screens"},
        {"id": "64a01f2", "name": "Audit Webhooks", "desc": "Check Stripe event handlers"},
        {"id": "64a01f3", "name": "Set up CDN Caching", "desc": "Configure Cloudflare page rules"}
    ]
    
    run_migration(sample_source_cards)

4. Disaster Recovery: What If You Lose Your Checkpoint File?

If someone deletes migration_checkpoint.json, running the script normally would create duplicates. Because our payload embeds *Migrated from Trello Card ID: {card_id}* in the task description (or alternatively in a ClickUp Custom Field), you can reconstruct your mapping with this recovery query:

# Retrieve all list tasks and extract the embedded ID from descriptions
GET https://api.clickup.com/api/v2/list/{list_id}/task?include_closed=true&subtasks=true
# Regex extraction in Python:
trello_id = re.search(r"Trello Card ID: ([a-zA-Z0-9]+)", task['description']).group(1)

5. People Also Ask: ClickUp API Essentials

Where can I find ClickUp API documentation and postman collections?

The official documentation is available at https://clickup.com/api. ClickUp maintains an active Postman Workspace where you can import the entire ClickUp API collection, configure your environment variables (token, team_id, list_id), and test payloads interactively before executing Python batches.

What is the difference between ClickUp API v2 and v3?

ClickUp API v2 is the stable, battle-tested workhorse handling all standard CRUD operations for tasks, lists, custom fields, and webhooks. ClickUp API v3 introduces streamlined pagination patterns, improved response schemas, and unified workspace endpoints. For high-volume migrations, v2 remains widely supported with maximum library stability.

How do ClickUp API pricing tiers affect rate limits?

ClickUp API rate limits are governed by workspace subscription tiers. Free and Unlimited plans are capped at 100 requests per minute. Business, Business Plus, and Enterprise tiers scale to 900–10,000 requests per minute. If you encounter frequent 429 errors despite exponential backoff, consult our guide on Fixing Trello to ClickUp Migration Failed Errors to tune your connection pools.

Written by Sethupathi S. Explore our complete automation series for real-world enterprise engineering workflows.