Python SDK

pip install decodo-sdk

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

export DECODO_API_KEY=your_api_key

# main.py
import os

from decodo import (
    DecodoClient,
    DecodoConfig,
    GoogleSearchParams,
    WebScrapingApiConfig,
)

client = DecodoClient(
    DecodoConfig(
        web_scraping_api=WebScrapingApiConfig(
            api_key=os.environ["DECODO_API_KEY"],
        ),
    )
)

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)

python -m decodo.codegen.codegen --out-dir ./decodo_generated

from decodo_generated.targets import GoogleSearchParams

from decodo import (
    DecodoClient,
    DecodoConfig,
    WebScrapingApiConfig,
)

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

# Basic authentication token (older plans)
client = DecodoClient(
    DecodoConfig(
        web_scraping_api=WebScrapingApiConfig(
            token="<basic_auth_token>",
        ),
    )
)

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)

from decodo import SearchRequest

result = client.web_scraping_api.search(
    SearchRequest(query="coffee shops", geo="United States", page=1)
)

for item in result.get("organic", []):
    print(item["rank"], item["title"], item["link"])

from decodo import MapRequest

result = client.web_scraping_api.map(
    MapRequest(url="https://www.example.com", limit=100, sitemap="include")
)

for link in result["results"][0]["content"]["links"]:
    print(link["url"])

from decodo import (
    CrawlError,
    CrawlRequest,
    CrawlTimeoutError,
)

try:
    pages = client.web_scraping_api.crawl(
        CrawlRequest(url="https://www.example.com", limit=20, max_depth=2),
        poll_interval=5,
        timeout=600,
    )
except CrawlTimeoutError as err:
    print(f"Still running, fetch it later with get_crawl_results({err.crawl_id!r})")
except CrawlError as err:
    print(f"Crawl {err.crawl_id} ended {err.status}: {err.reason}")
else:
    for page in pages:
        print(page["url"], page["status"], page.get("content", "")[:80])

started = client.web_scraping_api.start_crawl(
    CrawlRequest(url="https://www.example.com")
)
crawl_id = started["crawl_id"]

status = client.web_scraping_api.get_crawl_status(crawl_id)
print(status["status"], status["completed"], status["in_flight"])

pages = client.web_scraping_api.get_crawl_results(crawl_id)

client.web_scraping_api.cancel_crawl(crawl_id)

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 API key or token. Check your credentials in the 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}")