Building a Document Processing Pipeline with REST APIs

A technical deep-dive into designing document processing architectures using REST APIs. Learn how to chain extraction, summarization, and conversion into robust pipelines.

Modern applications don't just process text — they ingest, transform, analyze, and route documents through multi-step workflows. Whether you're building a RAG system, a document management platform, or an automated reporting tool, you need a document processing pipeline.

This guide walks through the architecture of building such a pipeline using REST APIs, with DocuMind API as the core processing engine. We'll cover synchronous vs asynchronous patterns, error handling, retry strategies, and real code.

What Is a Document Processing Pipeline?

A document processing pipeline is a sequence of operations that transform raw documents into structured, actionable data. Think of it as an assembly line for documents:

Each stage is independent, which means you can swap implementations, scale individual components, and handle failures gracefully.

The DocuMind API Endpoints

DocuMind provides three REST endpoints that map directly to common pipeline stages:

Pattern 1: Simple Sequential Pipeline

The simplest pattern processes documents through each stage sequentially. This works well for low-volume, real-time applications.

import requests
import json

class DocumentPipeline:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://api.tokenall.net.cn/documind/api/v1"
        self.headers = {"X-API-Key": api_key}
    
    def process(self, file_path):
        """Run a document through the full pipeline."""
        print(f"📄 Processing: {file_path}")
        
        # Stage 1: Extract text
        print("  [1/3] Extracting text...")
        text = self._extract(file_path)
        
        # Stage 2: Summarize
        print("  [2/3] Generating summary...")
        summary = self._summarize(file_path)
        
        # Stage 3: Convert to Markdown
        print("  [3/3] Converting to Markdown...")
        markdown = self._convert(file_path, "markdown")
        
        return {
            "source": file_path,
            "text": text,
            "summary": summary,
            "markdown": markdown
        }
    
    def _extract(self, file_path):
        with open(file_path, "rb") as f:
            resp = requests.post(
                f"{self.base_url}/extract",
                headers=self.headers,
                files={"file": f}
            )
        resp.raise_for_status()
        return resp.json()["text"]
    
    def _summarize(self, file_path):
        with open(file_path, "rb") as f:
            resp = requests.post(
                f"{self.base_url}/summarize",
                headers=self.headers,
                files={"file": f},
                data={"summary_length": "moderate"}
            )
        resp.raise_for_status()
        return resp.json()["summary"]
    
    def _convert(self, file_path, target_format):
        with open(file_path, "rb") as f:
            resp = requests.post(
                f"{self.base_url}/convert",
                headers=self.headers,
                files={"file": f},
                data={"target_format": target_format}
            )
        resp.raise_for_status()
        return resp.json()["content"]

# Usage
pipeline = DocumentPipeline("your_api_key")
result = pipeline.process("quarterly-report.pdf")

# Save results
with open("output.json", "w") as f:
    json.dump(result, f, indent=2)

Pattern 2: Resilient Pipeline with Retries

Production pipelines need to handle transient failures — network timeouts, rate limits, temporary server issues. Here's a more robust implementation:

import time
import hashlib
from functools import wraps
from pathlib import Path

def retry(max_attempts=3, base_delay=1.0, backoff=2.0):
    """Decorator for exponential backoff retry."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        # Rate limited - use Retry-After header
                        delay = int(e.response.headers.get("Retry-After", base_delay))
                    elif e.response.status_code >= 500 and attempt < max_attempts - 1:
                        delay = base_delay * (backoff ** attempt)
                    else:
                        raise
                    
                    print(f"    ⏳ Retry {attempt+1}/{max_attempts} after {delay:.1f}s")
                    time.sleep(delay)
            raise Exception(f"Failed after {max_attempts} attempts")
        return wrapper
    return decorator

class ResilientPipeline(DocumentPipeline):
    def __init__(self, api_key, cache_dir="./pipeline_cache"):
        super().__init__(api_key)
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)
    
    def _file_hash(self, file_path):
        """Generate MD5 hash for caching."""
        with open(file_path, "rb") as f:
            return hashlib.md5(f.read()).hexdigest()
    
    @retry(max_attempts=3)
    def _extract(self, file_path):
        return super()._extract(file_path)
    
    @retry(max_attempts=3)
    def _summarize(self, file_path):
        return super()._summarize(file_path)
    
    @retry(max_attempts=3)
    def _convert(self, file_path, target_format):
        return super()._convert(file_path, target_format)
    
    def process_with_cache(self, file_path):
        """Process with result caching to avoid redundant API calls."""
        file_hash = self._file_hash(file_path)
        cache_file = self.cache_dir / f"{file_hash}.json"
        
        if cache_file.exists():
            print(f"  ✓ Cache hit: {cache_file.name}")
            with open(cache_file) as f:
                return json.load(f)
        
        result = self.process(file_path)
        
        with open(cache_file, "w") as f:
            json.dump(result, f, indent=2)
        
        return result

Pattern 3: Batch Processing with Rate Limiting

For processing hundreds of documents, you need batch processing with built-in rate limiting:

import asyncio
import aiohttp
from datetime import datetime

class BatchProcessor:
    def __init__(self, api_key, max_concurrent=3, requests_per_minute=50):
        self.api_key = api_key
        self.base_url = "https://api.tokenall.net.cn/documind/api/v1"
        self.max_concurrent = max_concurrent
        self.min_interval = 60.0 / requests_per_minute
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_batch(self, file_paths, output_dir="./results"):
        """Process multiple files with concurrency control."""
        Path(output_dir).mkdir(exist_ok=True)
        
        async with aiohttp.ClientSession() as session:
            tasks = []
            for path in file_paths:
                tasks.append(self._process_one(session, path, output_dir))
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            success = sum(1 for r in results if not isinstance(r, Exception))
            failed = sum(1 for r in results if isinstance(r, Exception))
            print(f"\n✅ Batch complete: {success} succeeded, {failed} failed")
        
        return results
    
    async def _process_one(self, session, file_path, output_dir):
        async with self.semaphore:
            # Rate limiting
            await asyncio.sleep(self.min_interval)
            
            filename = Path(file_path).stem
            print(f"  Processing {filename}...")
            
            try:
                # Extract
                text = await self._api_call(session, "extract", file_path)
                
                # Summarize
                summary = await self._api_call(session, "summarize", file_path, 
                                               extra_data={"summary_length": "brief"})
                
                # Save result
                result = {
                    "file": filename,
                    "text": text.get("text", ""),
                    "summary": summary.get("summary", ""),
                    "processed_at": datetime.now().isoformat()
                }
                
                output_file = Path(output_dir) / f"{filename}.json"
                with open(output_file, "w") as f:
                    json.dump(result, f, indent=2)
                
                print(f"  ✓ {filename}")
                return result
                
            except Exception as e:
                print(f"  ✗ {filename}: {e}")
                return e
    
    async def _api_call(self, session, endpoint, file_path, extra_data=None):
        url = f"{self.base_url}/{endpoint}"
        data = aiohttp.FormData()
        data.add_field("file", open(file_path, "rb"), filename=Path(file_path).name)
        
        if extra_data:
            for key, value in extra_data.items():
                data.add_field(key, value)
        
        async with session.post(url, 
                                headers={"X-API-Key": self.api_key},
                                data=data) as resp:
            resp.raise_for_status()
            return await resp.json()

# Usage
async def main():
    processor = BatchProcessor("your_api_key", max_concurrent=3)
    files = list(Path("./documents/").glob("*.pdf"))
    await processor.process_batch(files)

asyncio.run(main())

Pattern 4: Event-Driven Pipeline with Webhooks

For production systems, an event-driven architecture gives you the most flexibility. Documents trigger events, and each pipeline stage processes independently:

from flask import Flask, request, jsonify
import threading
import uuid

app = Flask(__name__)
pipeline = ResilientPipeline("your_api_key")

# In-memory job store (use Redis/DB in production)
jobs = {}

@app.route("/api/upload", methods=["POST"])
def upload_document():
    """Accept document uploads and queue for processing."""
    if "file" not in request.files:
        return jsonify({"error": "No file provided"}), 400
    
    file = request.files["file"]
    job_id = str(uuid.uuid4())
    
    # Save uploaded file
    temp_path = f"/tmp/{job_id}.pdf"
    file.save(temp_path)
    
    # Create job record
    jobs[job_id] = {
        "id": job_id,
        "status": "processing",
        "filename": file.filename,
        "created_at": datetime.now().isoformat()
    }
    
    # Start async processing
    thread = threading.Thread(
        target=process_document_async,
        args=(job_id, temp_path)
    )
    thread.start()
    
    return jsonify({"job_id": job_id, "status": "processing"}), 202

def process_document_async(job_id, file_path):
    """Background processing thread."""
    try:
        result = pipeline.process_with_cache(file_path)
        jobs[job_id].update({
            "status": "completed",
            "result": result
        })
    except Exception as e:
        jobs[job_id].update({
            "status": "failed",
            "error": str(e)
        })

@app.route("/api/jobs/<job_id>", methods=["GET"])
def get_job_status(job_id):
    """Check processing status."""
    job = jobs.get(job_id)
    if not job:
        return jsonify({"error": "Job not found"}), 404
    return jsonify(job)

if __name__ == "__main__":
    app.run(port=8080)

Choosing the Right Pattern

Pipeline Architecture Best Practices

1. Idempotency — Cache results by file hash to avoid re-processing identical documents
2. Circuit breakers — If the API returns errors repeatedly, pause and alert rather than hammering the endpoint
3. Dead letter queues — Failed documents should be saved for manual review, not silently dropped
4. Observability — Log processing time, success rates, and quota usage for each pipeline run
5. Graceful degradation — If summarization fails, still return the extracted text

Getting Started

Ready to build your document processing pipeline? Start with DocuMind API:

1. Visit RapidAPI and search for "DocuMind"
2. Subscribe to the Free plan (100 requests/month)
3. Use the Simple Sequential pattern to prototype
4. Gradually add resilience patterns as you move to production

Published August 2, 2026 · TokenAll Blog