DEV Community

Elowen
Elowen

Posted on

Handle Timeouts, Empty Results, and Retries in SERP API Workflows

A SERP API workflow usually works well in a demo because the happy path is simple: send a query, get results, parse the response, move on.

The real workflow becomes more interesting when a request times out, the result is empty, or a retry creates duplicate output. If those states are not handled separately, the pipeline becomes hard to trust.

This post shows a small reliability pattern for workflows that call TalorData SERP API: use explicit timeouts, classify empty results, retry only when it makes sense, and log enough context for review.

The states to separate

Do not treat every non-happy path as the same failure.

At minimum, separate these states:

  • request succeeded with organic results
  • request succeeded but returned no organic results
  • request timed out
  • request failed before a usable response was returned
  • parsing failed after a response was received
  • workflow skipped the query intentionally

This separation matters because each state needs a different action.

An empty result may need review. A timeout may deserve a retry. A parsing issue probably needs a code fix. A skipped query should not be retried at all.

Basic request wrapper

Here is a small Python wrapper using requests.

import os
import time
from dataclasses import dataclass
from typing import Any

import requests


API_URL = "https://serpapi.talordata.net/serp/v1/request"


@dataclass
class SerpRunResult:
    status: str
    query: str
    attempts: int
    organic_count: int = 0
    data: dict[str, Any] | None = None
    note: str | None = None
Enter fullscreen mode Exit fullscreen mode

Use a placeholder token in docs and examples. In a real app, read the token from an environment variable or secret manager.

def request_serp(query: str, timeout_seconds: int = 30) -> dict[str, Any]:
    token = os.environ["TALORDATA_TOKEN"]

    response = requests.post(
        API_URL,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/x-www-form-urlencoded",
        },
        data={
            "engine": "google",
            "q": query,
            "num": "10",
            "json": "2",
        },
        timeout=timeout_seconds,
    )

    response.raise_for_status()
    return response.json()
Enter fullscreen mode Exit fullscreen mode

If you are documenting the header directly, keep it safe:

Authorization: Bearer <TALORDATA_TOKEN>
Enter fullscreen mode Exit fullscreen mode

Classify empty results

An empty result is not automatically the same as a failed request.

def classify_success(query: str, data: dict[str, Any], attempts: int) -> SerpRunResult:
    organic = data.get("organic", [])

    if not organic:
        return SerpRunResult(
            status="empty_result",
            query=query,
            attempts=attempts,
            organic_count=0,
            data=data,
            note="Request succeeded, but no organic results were available for this query.",
        )

    return SerpRunResult(
        status="succeeded",
        query=query,
        attempts=attempts,
        organic_count=len(organic),
        data=data,
    )
Enter fullscreen mode Exit fullscreen mode

This distinction helps downstream nodes decide what to do next.

A report may show empty results as a review item. A retry queue may ignore them unless the query is business-critical.

Add limited retry with backoff

Retries should be limited and visible.

def fetch_with_retry(query: str, max_attempts: int = 3) -> SerpRunResult:
    last_note = None

    for attempt in range(1, max_attempts + 1):
        try:
            data = request_serp(query)
            return classify_success(query, data, attempt)
        except requests.Timeout:
            last_note = "Request timed out."
        except requests.RequestException:
            last_note = "Request failed before a usable response was returned."
        except ValueError:
            return SerpRunResult(
                status="parse_failed",
                query=query,
                attempts=attempt,
                note="Response was received, but JSON parsing failed.",
            )

        if attempt < max_attempts:
            time.sleep(2 ** (attempt - 1))

    return SerpRunResult(
        status="failed_after_retries",
        query=query,
        attempts=max_attempts,
        note=last_note,
    )
Enter fullscreen mode Exit fullscreen mode

The key is not the exact backoff formula. The key is that retry behavior is bounded and recorded.

Decide what each status should do

A workflow becomes easier to maintain when each status has an action.

ACTIONS = {
    "succeeded": "store_results",
    "empty_result": "store_review_item",
    "parse_failed": "send_to_engineering_review",
    "failed_after_retries": "queue_for_later_review",
}
Enter fullscreen mode Exit fullscreen mode

For example:

  • succeeded: normalize and store organic results
  • empty_result: store the query, timestamp, and request settings for review
  • parse_failed: inspect the parser and response shape
  • failed_after_retries: preserve the failed query without blocking the whole run

Log the context

A useful log record should include:

query
status
attempts
organic_count
timestamp
request settings
next action
Enter fullscreen mode Exit fullscreen mode

If the workflow runs in batches, also include a run_id and query_id.

That makes it possible to answer later: did the workflow fail, return no results, or skip a query intentionally?

Final thought

Reliable search workflows do not retry everything blindly.

They classify what happened, retry only where retrying is useful, and preserve enough context for a human to inspect the edge cases.

If you want to test this reliability pattern with live Google results, TalorData gives new accounts 500 responses to build a small SERP API workflow before scaling it.

Top comments (0)