Add Retry and Backoff Around Search API Calls

A batch SERP collection job usually fails in boring ways.

One request times out.

One response returns a temporary error.

One keyword fails halfway through a run.

If your script stops there, the problem is not the SERP API call itself. The problem is that the job has no reliability wrapper around the call.

This post shows a small pattern for adding retry, backoff, timeout, and error logging around search API calls. The same pattern works whether your collection layer uses TalorData SERP API or another provider.

What retry should handle

Retry is useful for temporary failures, not every failure.

Good retry candidates:

• network timeout
• temporary server error
• connection reset
• rate-limit response if your provider documentation says retrying later is appropriate
• short-lived infrastructure issues

Bad retry candidates:

• invalid API key
• malformed query parameters
• unsupported location value
• missing required request field
• permanent validation error

Retrying permanent failures only makes the job slower and noisier.

A simple retry wrapper

This example is intentionally generic. Replace endpoint, auth, parameters, and retryable status codes with the actual API documentation for your provider.

import os
import random
import time
from typing import Any

import requests


SERP_API_ENDPOINT = os.environ["SERP_API_ENDPOINT"]
SERP_API_KEY = os.environ["SERP_API_KEY"]

RETRYABLE_STATUS_CODES = {429, 500, 502, 503, 504}


class SerpRequestError(Exception):
    pass


def fetch_serp_once(keyword: str, location: str) -> dict[str, Any]:
    response = requests.get(
        SERP_API_ENDPOINT,
        headers={"Authorization": f"Bearer {SERP_API_KEY}"},
        params={
            "q": keyword,
            "location": location,
        },
        timeout=30,
    )

    if response.status_code in RETRYABLE_STATUS_CODES:
        raise SerpRequestError(f"Retryable status code: {response.status_code}")

    response.raise_for_status()
    return response.json()

Add exponential backoff

Do not retry immediately in a tight loop.

Use backoff with a small amount of jitter:

def backoff_seconds(attempt: int, base: float = 1.0, cap: float = 30.0) -> float:
    delay = min(cap, base * (2 ** attempt))
    jitter = random.uniform(0, 0.5 * delay)
    return delay + jitter


def fetch_serp_with_retry(keyword: str, location: str, max_attempts: int = 4) -> dict[str, Any]:
    last_error: Exception | None = None

    for attempt in range(max_attempts):
        try:
            return fetch_serp_once(keyword, location)
        except (requests.Timeout, requests.ConnectionError, SerpRequestError) as error:
            last_error = error

            if attempt == max_attempts - 1:
                break

            time.sleep(backoff_seconds(attempt))

    raise SerpRequestError(
        f"SERP request failed after {max_attempts} attempts for keyword={keyword!r} location={location!r}"
    ) from last_error

This keeps temporary failures from breaking the full batch.

Log failures instead of hiding them

Retry should not make failures invisible.

If a keyword fails after all attempts, write a failure record:

import csv
from datetime import datetime, timezone
from pathlib import Path


FAILURE_LOG = Path("serp_failures.csv")


def log_failure(keyword: str, location: str, error: Exception) -> None:
    file_exists = FAILURE_LOG.exists()

    with FAILURE_LOG.open("a", newline="", encoding="utf-8") as file:
        writer = csv.DictWriter(
            file,
            fieldnames=["failed_at", "keyword", "location", "error"],
        )

        if not file_exists:
            writer.writeheader()

        writer.writerow(
            {
                "failed_at": datetime.now(timezone.utc).isoformat(),
                "keyword": keyword,
                "location": location,
                "error": str(error),
            }
        )

The failure log helps you rerun only the missing pieces later.

Use retry inside a batch job

def collect_batch(configs: list[dict[str, str]]) -> list[dict[str, Any]]:
    snapshots = []

    for config in configs:
        keyword = config["keyword"]
        location = config["location"]

        try:
            raw = fetch_serp_with_retry(keyword, location)
            snapshots.append(
                {
                    "keyword": keyword,
                    "location": location,
                    "raw": raw,
                }
            )
        except Exception as error:
            log_failure(keyword, location, error)

    return snapshots

This job completes even if a few keyword-location pairs fail.

That is usually better than losing the entire run.

What to add later

Once the basic retry wrapper works, add:

• request IDs or job IDs
• structured logs
• retry count per request
• failure reason categories
• rerun command for failed keywords
• dashboard or report section for missing snapshots
• alert only when failure rate crosses a threshold

Do not start with a complex job system if a simple wrapper and failure log solves the first problem.

Where the SERP API fits

The SERP API provides structured search result data.

Your reliability layer makes collection repeatable enough to trust.

A tool like TalorData can provide the search data layer, while your application owns retry policy, logging, reruns, and reporting around failed jobs.

Final note

If you run scheduled SERP collection, what do you track first: retry count, failed keywords, missing snapshots, or total failure rate?