Python SDK

pip install decodo-sdk

mkdir scrape-with-decodo
cd scrape-with-decodo
pip install decodo-sdk
touch main.py

export DECODO_TOKEN=your_token

# main.py
from decodo import (
    DecodoClient,
    DecodoConfig,
    GoogleSearchParams,
    WebScrapingApiConfig,
)

client = DecodoClient(
    DecodoConfig(
        web_scraping_api=WebScrapingApiConfig(
            token=os.environ["DECODO_TOKEN"],
        ),
    )
)

result = client.web_scraping_api.scrape(
    GoogleSearchParams(
        query="coffee shops",
        geo="United States",
        parse=True,
    )
)

print(result)

python main.py

result = client.web_scraping_api.scrape({
    "target": "google_search",
    "query": "coffee shops",
    "geo": "United States",
    "parse": True,
})
print(result)

from decodo import (
    DecodoClient,
    DecodoConfig,
    WebScrapingApiConfig,
)

client = DecodoClient(
    DecodoConfig(
        web_scraping_api=WebScrapingApiConfig(
            token="<token>",
        ),
        timeout_ms=120_000,  # optional, request timeout in ms (default: 180000)
    )
)

from decodo import AmazonProductParams

result = client.web_scraping_api.scrape(
    AmazonProductParams(
        query="B09H74FXNW",
        parse=True,
    )
)
print(result)

import time
from decodo import GoogleSearchParams

task = client.web_scraping_api.scrape_async(
    GoogleSearchParams(query="laptop reviews", parse=True)
)
task_id = task["id"]

while True:
    status = client.web_scraping_api.get_status(task_id)["status"]
    if status == "done":
        break
    if status == "faulted":
        raise RuntimeError(f"Scraping task {task_id} failed.")
    time.sleep(2)

result = client.web_scraping_api.get_results(task_id)
print(result)

import time
from decodo import GoogleSearchBatchParams

batch = client.web_scraping_api.scrape_batch(
    GoogleSearchBatchParams(query=["coffee", "tea", "juice"], parse=True)
)

for task in batch["queries"]:
    task_id = task["id"]
    while True:
        status = client.web_scraping_api.get_status(task_id)["status"]
        if status == "done":
            break
        if status == "faulted":
            raise RuntimeError(f"Scraping task {task_id} failed.")
        time.sleep(2)
    result = client.web_scraping_api.get_results(task_id)
    print(result)

import time
from decodo import (
    AuthenticationError,
    DecodoError,
    GoogleSearchParams,
    RateLimitError,
    TimeoutError,
    ValidationError,
)

params = GoogleSearchParams(query="coffee shops", geo="United States", parse=True)

try:
    response = client.web_scraping_api.scrape(params)
except AuthenticationError:
    raise SystemExit("Invalid token. Check the Basic Auth credentials in your dashboard.")
except RateLimitError:
    time.sleep(5)
    response = client.web_scraping_api.scrape(params)
except ValidationError as err:
    print("Payload rejected:", err.errors or err)
except TimeoutError:
    print("Timed out. Increase timeout_ms or switch to scrape_async for slow targets.")
except DecodoError as err:
    print(f"Request failed with {err.status_code}: {err}")