Python client for Google Shopping and Google Search data

Step by step

1. 01Add the client

   Save this as crawlerdata_client.py. It retries network errors, 429 and 5xx responses with backoff, waiting until X-RateLimit-Reset when the API sends it. 400, 401 and 402 raise straight away, because retrying will not fix them.

   live() returns the result of a live request. submit() queues a job on the fast or regular queue and returns its ID, and wait() polls GET /v1/jobs/{id} until the job completes or fails.

   """Minimal CrawlerData client: live requests, async jobs with polling, retries. Needs requests."""
   import os
   import random
   import time
   
   import requests
   
   API = "https://api.crawlerdata.com/v1"
   
   
   class CrawlerDataError(Exception):
       pass
   
   
   class CrawlerData:
       def __init__(self, api_key=None, timeout=200, max_retries=4):
           self.session = requests.Session()
           self.session.headers["Authorization"] = f"Bearer {api_key or os.environ['CRAWLERDATA_API_KEY']}"
           self.timeout = timeout
           self.max_retries = max_retries
   
       def _request(self, method, path, **kwargs):
           for attempt in range(self.max_retries + 1):
               try:
                   res = self.session.request(method, API + path, timeout=self.timeout, **kwargs)
               except (requests.ConnectionError, requests.Timeout):
                   if attempt == self.max_retries:
                       raise
                   time.sleep(2 ** attempt + random.random())
                   continue
               # 429 = rate limited, 5xx = temporary. Wait and try again.
               if res.status_code == 429 or res.status_code >= 500:
                   if attempt == self.max_retries:
                       break
                   reset = res.headers.get("X-RateLimit-Reset")
                   wait = max(1, int(reset) - int(time.time())) if reset and reset.isdigit() else 2 ** attempt
                   time.sleep(min(wait, 60) + random.random())
                   continue
               break
           if res.status_code >= 400:
               # 400 bad params, 401 bad key, 402 out of credits: retrying will not help.
               raise CrawlerDataError(f"HTTP {res.status_code}: {res.text[:300]}")
           return res.json()
   
       def run(self, endpoint, params, queue="live", postback_url=None):
           """Submit a job. On the live queue the response holds the result."""
           body = {"endpoint": endpoint, "params": params, "queue": queue}
           if postback_url:
               body["postbackUrl"] = postback_url
           return self._request("POST", "/jobs", json=body)
   
       def get_job(self, job_id):
           return self._request("GET", f"/jobs/{job_id}")
   
       def live(self, endpoint, params):
           """Run a live request and return the result, or raise if the job failed."""
           data = self.run(endpoint, params, queue="live")
           if data["job"]["status"] != "completed":
               raise CrawlerDataError(f"job {data['job']['id']} failed: {data['job']['error']}")
           return data["result"]
   
       def submit(self, endpoint, params, queue="fast", postback_url=None):
           """Queue a job on fast or regular and return its id."""
           return self.run(endpoint, params, queue=queue, postback_url=postback_url)["job"]["id"]
   
       def wait(self, job_id, interval=5, max_wait=600):
           """Poll a job until it completes or fails."""
           deadline = time.time() + max_wait
           while time.time() < deadline:
               data = self.get_job(job_id)
               status = data["job"]["status"]
               if status == "completed":
                   return data["result"]
               if status == "failed":
                   raise CrawlerDataError(f"job {job_id} failed: {data['job']['error']}")
               time.sleep(interval)
           raise TimeoutError(f"job {job_id} still running after {max_wait}s")

2. 02Use it

   Use live for one-off lookups. For batches, submit every job first and then wait for each one, so the jobs run in parallel instead of one after another.

   from crawlerdata_client import CrawlerData
   
   cd = CrawlerData()  # reads CRAWLERDATA_API_KEY
   
   # Live: the result comes back in the same response.
   result = cd.live("google-search", {"query": "best crm software", "country": "us"})
   for r in result["organic_results"][:3]:
       print(r["position"], r["title"], r["link"])
   
   # Async: submit a batch on the fast queue, then poll each job.
   ids = [cd.submit("google-autocomplete", {"query": q}) for q in ["running sho", "trail runn"]]
   for job_id in ids:
       print([s["value"] for s in cd.wait(job_id)["suggestions"][:3]])

3. 03Receive results by webhook

   Pass postback_url to submit() and the result is POSTed to that URL when the job finishes. The payload has jobId, endpoint, status, data (the result), credits, latencyMs and timestamp, and the request carries an X-CrawlerData-Job-Id header.

   If you set a webhook secret in the dashboard (API Access, Webhook), each delivery has an X-CrawlerData-Signature header: the hex HMAC-SHA256 of the raw body with your secret. Verify it against the raw bytes, before parsing the JSON.

   Answer with a 2xx within 10 seconds, and use GET /v1/jobs/{id} as a fallback for any job whose webhook you did not receive.

   """Receive CrawlerData webhooks (fast and regular queues). Standard library only."""
   import hashlib
   import hmac
   import json
   import os
   from http.server import BaseHTTPRequestHandler, HTTPServer
   
   # The webhook secret you set in the dashboard (API Access > Webhook). Leave unset to skip the check.
   SECRET = os.environ.get("CRAWLERDATA_WEBHOOK_SECRET", "")
   
   
   def valid_signature(body: bytes, signature: str) -> bool:
       expected = hmac.new(SECRET.encode(), body, hashlib.sha256).hexdigest()
       return hmac.compare_digest(expected, signature or "")
   
   
   class Handler(BaseHTTPRequestHandler):
       def do_POST(self):
           body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
           if SECRET and not valid_signature(body, self.headers.get("X-CrawlerData-Signature", "")):
               self.send_response(401)
               self.end_headers()
               return
           payload = json.loads(body)
           # Answer with a 2xx within 10 seconds, then do the slow work. Poll GET /v1/jobs/{id} for any job you missed.
           self.send_response(204)
           self.end_headers()
           if payload["status"] == "completed":
               handle_result(payload["jobId"], payload["endpoint"], payload["data"])
           else:
               print("job failed", payload["jobId"], payload.get("error"))
   
   
   def handle_result(job_id, endpoint, result):
       # Store it, queue it, etc. Keep heavy work out of the request handler.
       print(job_id, endpoint, json.dumps(result)[:200])
   
   
   if __name__ == "__main__":
       HTTPServer(("0.0.0.0", int(os.environ.get("PORT", 8000))), Handler).serve_forever()

If something goes wrong

401 INVALID_API_KEY

The key is missing, mistyped or revoked. Send it as "Authorization: Bearer YOUR_API_KEY" (with the word Bearer and one space) and check the key is active under API Access in the dashboard.

402 INSUFFICIENT_CREDITS

Your balance is too low for the request. Buy a credit pack in the dashboard. Requests that fail for other reasons are not charged, so a 402 never costs credits.

429 RATE_LIMITED

Each API key allows 100 requests per minute and 20 per second. Wait until the time in the X-RateLimit-Reset header (a Unix timestamp) and retry, or spread requests out.

Timeouts on live requests

A live request returns when the result is ready. Search requests usually take a few seconds; a Google Shopping sellers request collects every seller for the product, so allow more time. Set the client timeout to about 200 seconds, or use the fast queue and poll the job.

job.status is "failed"

The HTTP request worked but the job did not; job.error says why. Failed jobs cost 0 credits, so you can retry them.

Webhooks never arrive

The URL must be public over HTTPS or HTTP and answer within 10 seconds with a 2xx. Poll GET /v1/jobs/{id} for any job you have not heard back about.

Request and response schemas: OpenAPI 3.1 document. Still stuck? Email contact@crawlerdata.com.

APIs you can call

Solutions built on it

Other integrations