Robust Pipeline – Successfully Handling 500 Calls with Exceptions, Retries, and Circuit Breakers

After Completing This Topic

By combining the concepts of exceptions, try-except, and error handling learned in the textbook, you can build a tool to ensure that a pipeline that makes a large number of calls to external APIs completes successfully, even if some calls fail. This is a practical pattern for improving the success rate from 95% to 99.9%.

This article is a general educational example. In real-world scenarios, libraries like tenacity, backoff, and resilience4j are used.

"The next morning, I opened the logs" — The Pitfalls of a Naive Pipeline

Let's say you ran a script overnight that sequentially calls NCBI BLAST for 500 genes.

python

def naive_pipeline(gene_ids: list[str]) -> list[dict]:
    results = []
    for gene_id in gene_ids:
        response = call_blast_api(gene_id)
        results.append(response)
    return results

The next morning, you open the logs and see this:

text

[00:03] gene 1: OK
[00:05] gene 2: OK
...
[01:23] gene 27: TIMEOUT
Traceback (most recent call last):
  ...
requests.exceptions.Timeout

The entire script stopped at the 27th call, and the remaining 473 calls were never even attempted. A single failure killed the entire run.

The problems with this approach:

Problem 1: No failure isolation. A single error brings everything down.

Problem 2: No retries. Network glitches are often transient. Waiting a bit and trying again will often succeed.

Problem 3: No progress saving. The 26 successful results are lost in memory when it crashes. We start from scratch again.

The real approach is a robust pipeline pattern. Isolate failures for each call, retry when necessary, and implement circuit breakers for persistent failures. Save successful results immediately.

From Black Box to Components

Component 1: Isolating Failures with Try-Except

python

def robust_pipeline_basic(gene_ids: list[str]) -> tuple[list[dict], list[dict]]:
    results = []
    errors = []
    
    for gene_id in gene_ids:
        try:
            response = call_blast_api(gene_id)
            results.append({"gene_id": gene_id, "data": response})
        except Exception as e:
            errors.append({"gene_id": gene_id, "error": type(e).__name__, "message": str(e)})
    
    return results, errors

Key Idea: Failed calls are isolated into the errors list, and the next iteration continues. Even if some of the 500 calls fail, the rest will complete.

Caution: except Exception catches all exceptions. This should only be used at the top level of the pipeline. If you want to catch specific errors, do so explicitly.

python

try:
    response = call_blast_api(gene_id)
except requests.exceptions.Timeout:
    # Retriable error
    pass
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 429:
        # Rate limit -> retriable
        pass
    else:
        # Other 4xx -> no point in retrying
        raise

Component 2: Exponential Backoff Retries

Transient errors are often resolved by retrying. However, retrying immediately puts a strain on the server. Instead, increase the interval between each retry exponentially.

python

import time
import random


def with_retry(func, max_attempts: int = 5, base_delay: float = 1.0):
    """
    Calls func, retrying with exponential backoff if it fails.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as e:
            if attempt == max_attempts:
                raise
            
            delay = base_delay * (2 ** (attempt - 1))
            jitter = random.uniform(0, delay * 0.1)
            time.sleep(delay + jitter)

Intervals: 1 second, 2 seconds, 4 seconds, 8 seconds, 16 seconds. With 5 retries, it waits for a maximum of 31 seconds.

Why add Jitter (random variation)? If multiple clients fail at the same time and retry at exactly the same time, the server will crash again. Adding a small random variation ensures that each client retries at a slightly different time.

Component 3: Circuit Breaker

If the server is down for an extended period, retrying is pointless. If the number of consecutive failures exceeds a threshold, temporarily stop calling the service altogether.

python

from dataclasses import dataclass
from enum import Enum


class CircuitState(Enum):
    CLOSED = "closed"      # Normal - allow calls
    OPEN = "open"          # Tripped - fail immediately
    HALF_OPEN = "half_open"  # Recovery attempt - test call


@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: float = 60.0
    state: CircuitState = CircuitState.CLOSED
    failure_count: int = 0
    last_failure_time: float = 0.0
    
    def call(self, func):
        now = time.time()
        
        if self.state == CircuitState.OPEN:
            if now - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
            else:
                raise Exception("Circuit breaker OPEN")
        
        try:
            result = func()
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
            return result
        except Exception:
            self.failure_count += 1
            self.last_failure_time = now
            if self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN
            raise

Three states:

CLOSED: Normal. Allow calls.
OPEN: Tripped. Return failure immediately (do not actually call the service).
HALF_OPEN: Recovery attempt. Use a single call to test the server's status.

Pipeline Assembly

Combine the three components.

python

def robust_pipeline_full(
    gene_ids: list[str],
    output_path: str,
    checkpoint_every: int = 50
) -> dict:
    breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)
    results = []
    errors = []
    
    # Load existing checkpoint
    already_done = load_checkpoint(output_path)
    
    for i, gene_id in enumerate(gene_ids):
        if gene_id in already_done:
            continue
        
        try:
            def call():
                return breaker.call(lambda: call_blast_api(gene_id))
            
            response = with_retry(call, max_attempts=3, base_delay=2.0)
            results.append({"gene_id": gene_id, "data": response})
        except Exception as e:
            errors.append({
                "gene_id": gene_id,
                "error": type(e).__name__,
                "message": str(e)
            })
        
        if (i + 1) % checkpoint_every == 0:
            save_checkpoint(output_path, results)
            print(f"Progress: {i+1}/{len(gene_ids)} — checkpoint saved")
    
    save_checkpoint(output_path, results)
    return {
        "total": len(gene_ids),
        "success": len(results),
        "failed": len(errors),
        "errors": errors
    }


def save_checkpoint(path: str, results: list) -> None:
    import json
    with open(path, "w") as f:
        json.dump(results, f, indent=2)


def load_checkpoint(path: str) -> set[str]:
    import json
    from pathlib import Path
    if not Path(path).exists():
        return set()
    with open(path) as f:
        data = json.load(f)
    return {r["gene_id"] for r in data}

Now, if a crash occurs, restarting the process will skip already processed genes and only attempt the remaining ones.

Fading — Three Blanks You Need to Fill

Blank 1: Exception-Specific Retry Policy

Only certain exceptions should be retried. For example, a 400 Bad Request is pointless to retry.

python

def with_retry_smart(func, max_attempts: int = 5):
    retryable = (
        requests.exceptions.Timeout,
        requests.exceptions.ConnectionError,
    )
    
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except retryable as e:
            # TODO: Retry logic (backoff + jitter)
            pass
        except requests.exceptions.HTTPError as e:
            # TODO: Check status_code and retry if 429 or 5xx
            # Otherwise, raise immediately
            pass

Hint: if e.response.status_code == 429 or 500 <= e.response.status_code < 600: sleep(delay); continue; else: raise.

Blank 2: Failure Report

Analyze the failed gene_ids and categorize them by cause.

python

def analyze_failures(errors: list[dict]) -> dict:
    """
    Groups the list of errors by type.
    Returns: {"Timeout": 5, "HTTPError_500": 3, ...}
    """
    from collections import Counter
    
    # TODO: Count by the "error" field of each error
    # Further categorize HTTPError by status_code
    pass

Hint: counter = Counter(); for e in errors: key = e["error"]; if "status" in e["message"]: key = f"{key}_{parse_status}"; counter[key] += 1.

Blank 3: Retry Only the Failed Parts on Re-execution

Save both successes and failures to a checkpoint, and retry only the failed ones on re-execution.

python

def retry_failed_only(
    gene_ids: list[str],
    output_path: str,
    error_log_path: str
) -> None:
    # TODO 1: Load from error_log_path
    # TODO 2: Extract only the failed gene_ids
    # TODO 3: Re-execute robust_pipeline_full (with the filtered list)
    pass

Reflection — Differences from Production-Ready Resilience Tools

tenacity library: A Python standard retry tool. Offers a much cleaner syntax using decorators.

python

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, max=60))
def call_api(gene_id):
    return requests.get(f"...").json()

Kubernetes readiness probes: In production deployments, the application itself monitors its own state. Traffic is rerouted when issues are detected.

Netflix Hystrix / resilience4j: Sophisticated circuit breakers in the Java ecosystem. Combines patterns such as Bulkhead, TimeLimiter, and RateLimiter.

Dead Letter Queue: Failed requests are stored in a separate queue for later manual or automatic reprocessing. A standard feature of AWS SQS and RabbitMQ.

Alternatives to Distributed Transactions: In production, systems are designed to be resilient to failures and recoverable (saga pattern, event sourcing). Failure is treated as a normal case rather than an exception.

Extension Project

1. Rewrite with Tenacity: Replace your backoff logic with tenacity decorators.

2. Parallel Processing: Use concurrent.futures.ThreadPoolExecutor to execute multiple calls concurrently. The circuit breaker is shared across multiple threads.

3. Dashboard: Display progress, success rate, and circuit breaker status in real time. If using Streamlit, this can be done in 30 minutes.

4. Notifications: Send notifications to Slack/Telegram when the failure rate exceeds a threshold.

Component Map for This Section

[F] Exceptions: Python's hierarchical exception system. Inherit from Exception and catch specific types.
[F] try-except: Isolate and recover from failures. Access error information using except ... as e.
[F] Error Handling: A unified pattern for retries, backoff, and circuit breakers.
[W] File I/O: Checkpoint saving and loading (a complete script will be provided).

[F] = You implement this yourself / [W] = A complete code example will be provided.

BioPlayground

A robust pipeline successfully handled 500 calls, incorporating exception handling, retries, and circuit breakers.

Robust Pipeline – Successfully Handling 500 Calls with Exceptions, Retries, and Circuit Breakers

After Completing This Topic

"The next morning, I opened the logs" — The Pitfalls of a Naive Pipeline

From Black Box to Components

Component 1: Isolating Failures with Try-Except

Component 2: Exponential Backoff Retries

Component 3: Circuit Breaker

Pipeline Assembly

Fading — Three Blanks You Need to Fill

Blank 1: Exception-Specific Retry Policy

Blank 2: Failure Report

Blank 3: Retry Only the Failed Parts on Re-execution

Reflection — Differences from Production-Ready Resilience Tools

Extension Project

Component Map for This Section

💬 Questions & Comments