Production-Grade Bulk API Processing with Python

Escaping Postman: A Pattern for High-Volume API Processing

UI-based API clients like Postman are excellent for exploration and single-request debugging. However, they are fundamentally ill-suited for high-volume operations. When attempting to process datasets exceeding 1,000 records, users inevitably encounter memory constraints, UI instability, and the inability to stream data directly to the local file system.

For operations requiring 10,000+ requests (e.g., data migration, bulk ID lookups, or status checks), a dedicated script is the only viable production solution.

The Architecture

A robust bulk-processing script must solve three specific engineering challenges that UI clients cannot:

1. Atomic Persistence: Data must be written to disk immediately following each request. Storing results in memory until the batch completes introduces a single point of failure where a crash at 99% completion results in 100% data loss.
2. Concurrency Control: Serial processing is too slow for large datasets. The solution requires a thread pool to manage parallel network I/O without overwhelming the local CPU or the remote server.
3. Input Resilience: CSVs generated by business tools (like Excel) often contain Byte Order Marks (BOM) or inconsistent encodings that break standard parsers.

The Generic Solution

The following Python script implements a reusable pattern for these requirements. It uses ThreadPoolExecutor for concurrency and threading.Lock to ensure thread-safe file I/O.

bulk_api_runner.py
This script is designed as a template. To use it, simply modify the process_request function to match your specific API payload.

import requests
import csv
import time
import threading
import sys
import argparse
from concurrent.futures import ThreadPoolExecutor

# --- CONFIGURATION ---
TARGET_URL = "https://api.yourservice.com/v1/resource"  # <--- SET URL
API_KEY = "YOUR_API_KEY" # <--- SET API KEY
INPUT_COLUMN = "email" # <--- The header name in your CSV to use as the key

# PERFORMANCE SETTINGS
MAX_WORKERS = 5       # Concurrency level. Adjust based on API rate limits.
DELAY_SECONDS = 0.5   # Sleep time per thread to smooth out traffic spikes.
MAX_RETRIES = 1       # Retry attempts for 5xx errors or timeouts.

# GLOBALS
lock = threading.Lock()
global_counter = 0

def process_request(identifier):
    """
    Executes a single API request and returns the result row.
    Customize the payload logic here.
    """
    global global_counter
    if not identifier: return
    identifier = identifier.strip()

    headers = {
        'Content-Type': 'application/json',
        'x-api-key': API_KEY
    }
    
    # --- CUSTOMIZE PAYLOAD HERE ---
    payload = {"query": identifier}
    
    final_status = "fail"
    final_result = "N/A"
    final_error = ""

    # --- RETRY LOOP ---
    for attempt in range(MAX_RETRIES + 1):
        try:
            response = requests.post(TARGET_URL, json=payload, headers=headers, timeout=10)
            
            # --- CUSTOMIZE SUCCESS LOGIC HERE ---
            if response.status_code == 200:
                data = response.json()
                result_value = data.get('id') # Extract the value you need
                
                if result_value:
                    final_status = "success"
                    final_result = result_value
                    final_error = ""
                    break 
                else:
                    final_error = "200 OK but expected data missing"
            else:
                final_error = f"HTTP {response.status_code}"

        except Exception as e:
            final_error = str(e)
        
        # Simple backoff
        if attempt < MAX_RETRIES:
            time.sleep(2)
    
    # --- ATOMIC WRITE (THREAD SAFE) ---
    with lock:
        # Open/Close file on every write to ensure data safety during crashes
        with open(args.output, 'a', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow([identifier, final_result, final_status, final_error])
        
        # CLI Feedback
        if final_status == "fail":
            sys.stdout.write(f"\n[!] Fail: {identifier} -> {final_error}\n")
        
        sys.stdout.write('.')
        global_counter += 1
        
        if global_counter % 50 == 0:
            sys.stdout.write(f"\n{global_counter} requests\n")
        
        sys.stdout.flush()

    time.sleep(DELAY_SECONDS)

def main():
    global args

    # --- ARGUMENT PARSING ---
    parser = argparse.ArgumentParser(description="Generic Bulk API Processor")
    parser.add_argument("input_file", help="Path to source CSV")
    parser.add_argument("-o", "--output", default="output_results.csv", help="Output CSV name")
    args = parser.parse_args()

    # --- INITIALIZE OUTPUT ---
    try:
        with open(args.output, 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(['input_key', 'result_value', 'status', 'error_detail'])
    except IOError as e:
        print(f"❌ Error: Could not create output file. {e}")
        return

    # --- ROBUST INPUT READING ---
    items_to_process = []
    print(f"Reading from: {args.input_file}")
    
    try:
        # 'utf-8-sig' handles Excel BOM (Byte Order Mark) automatically
        with open(args.input_file, 'r', encoding='utf-8-sig') as f:
            reader = csv.DictReader(f)
            
            # Case-insensitive header lookup
            headers = reader.fieldnames if reader.fieldnames else []
            target_key = next((h for h in headers if h.strip().lower() == INPUT_COLUMN.lower()), None)
            
            if not target_key:
                print(f"❌ Error: Could not find column '{INPUT_COLUMN}' in {headers}")
                return

            for row in reader:
                if row[target_key]:
                    items_to_process.append(row[target_key])
                    
    except FileNotFoundError:
        print(f"❌ Error: Input file '{args.input_file}' not found.")
        return

    print(f"Processing {len(items_to_process)} records...")
    print("-" * 50)
    
    # --- EXECUTION POOL ---
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        executor.map(process_request, items_to_process)

    print("\n" + "-" * 50)
    print(f"Done! Results saved to {args.output}")

if __name__ == "__main__":
    main()

Implementation Details

Handling Excel/BOM (utf-8-sig)

CSVs exported from Excel often include a Byte Order Mark (BOM) at the beginning of the file. To a standard UTF-8 parser, the first header column will appear as \ufeffcolumn_name rather than column_name. The script uses encoding='utf-8-sig' to automatically detect and strip this mark, preventing KeyError exceptions during header parsing.

Concurrency and Thread Safety

The script utilizes ThreadPoolExecutor to handle network latency. Because multiple threads operate simultaneously, file I/O must be synchronized. The with lock: block ensures that lines are written to the CSV atomically. Without this lock, race conditions would cause interleaved characters and corrupted output lines.

Immediate Persistence Strategy

The script opens and closes the file handle within the locked write block (with open(...)). While this incurs a minor I/O overhead compared to keeping the file open globally, it provides strict data safety. If the process is terminated (SIGINT or crash) at request #5,000, all previous records are guaranteed to be flushed to disk.

Usage

This workflow assumes a standard Python environment. Using a virtual environment is recommended to isolate dependencies.

# 1. Setup Environment
python3 -m venv venv
source venv/bin/activate
pip install requests

# 2. Run
python bulk_api_runner.py source_data.csv -o final_results.csv