Build a local business lead list from Google Maps in ~60 lines of Python

Extracto — Sat, 24 Oct 2026 19:08:30 +0000

Every sales, local-SEO or market-research project eventually needs the same thing: a spreadsheet of businesses in a city — name, phone, website, rating, address. Copy-pasting from Google Maps doesn't scale, and maintaining your own headless-browser scraper means fighting scrolling, rate limits and layout changes.

In this post we'll build a small Python script that:

estimates the cost of a run before launching it,
queues a Google Maps scraping job with an idempotency key (so network retries never double-charge),
polls the job politely until it finishes,
pages through the results with a cursor,
writes a deduplicated CSV ready for your CRM.

I'll use the Extracto HTTP API because it's a plain REST API (no SDK needed), but the queue → poll → paginate pattern applies to almost any async scraping or data API.

Setup

You need Python 3.9+ and requests:

pip install requests
export EXTRACTO_API_KEY="ext_your_api_key"

Every request carries the key in the X-API-KEY header. Keep it on the server — never in browser code or a public repo.

import csv, os, time, uuid
import requests

BASE = "https://extracto.cloud/api/v1"
PARSER = "google-maps-leads-actor"
HEADERS = {
    "X-API-KEY": os.environ["EXTRACTO_API_KEY"],
    "Content-Type": "application/json",
    "Accept": "application/json",
}

1. Describe what you want

The input is just JSON. One search per line — business type plus a location works best:

job_input = {
    "searchStrings": [
        "Dentist Austin TX USA",
        "Dental clinic Round Rock TX USA",
    ],
    "maxPlaces": 200,          # hard cap across the whole task
    "includeEmails": True,     # look for public emails on business websites
    "includeDetails": False,
    "includeReviews": False,
    "scrapeMode": "fast",
    "language": "en",
}

Two tips:

Always bound maxPlaces. 0 means "everything available", which is rarely what you want on a first run.
Options like website email checks or reviews cost extra, so turn on only what you'll actually use.

2. Estimate before you spend

The /estimate endpoint prices the exact same body without starting any scraping:

def estimate(inp):
    r = requests.post(f"{BASE}/parsers/{PARSER}/estimate",
                      json={"input": inp}, headers=HEADERS, timeout=30)
    r.raise_for_status()
    return r.json()

print(estimate(job_input))

For larger runs the API asks you to confirm a budget: when the estimate comes back with confirmation_required: true, pass the returned quote_token plus a max_budget_usd in your launch body. Final charges can never exceed that maximum, which is a nice safety net for scheduled jobs.

3. Queue the job — idempotently

Long scraping jobs shouldn't hold an HTTP connection open, so we use the queue endpoint and get a run ID back immediately (HTTP 202).

The important detail is the Idempotency-Key header. Generate one fresh key per intentional task and reuse it only when retrying that same task. If your connection drops after the server accepted the job, retrying with the same key returns the existing run instead of launching (and paying for) a second one.

def queue(inp, key):
    h = {**HEADERS, "Idempotency-Key": key}
    for attempt in range(5):
        try:
            r = requests.post(f"{BASE}/parsers/{PARSER}/queue",
                              json={"input": inp}, headers=h, timeout=60)
        except requests.ConnectionError:
            time.sleep(2 ** attempt)
            continue                      # same key -> safe to retry
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", 10)))
            continue
        r.raise_for_status()
        return r.json()["run"]["id"]
    raise RuntimeError("could not queue the job")

run_id = queue(job_input, key=f"maps-dentists-{uuid.uuid4()}")
print("queued:", run_id)

4. Poll politely

Check status roughly every five seconds while the run is queued or running, and stop at a terminal state. Note that a failed run still answers the status request with HTTP 200 — you have to look at run.status, not just the HTTP code.

TERMINAL = {"ok", "partial", "cancelled", "error"}

def wait(run_id):
    while True:
        r = requests.get(f"{BASE}/parser-runs/{run_id}",
                         params={"include_result": 0},
                         headers=HEADERS, timeout=30)
        r.raise_for_status()
        run = r.json()["run"]
        if run["status"] in TERMINAL:
            return run
        time.sleep(5)

run = wait(run_id)
if run["status"] == "error":
    raise SystemExit(f"run failed: {run['error']}")

include_result=0 keeps status checks lightweight — we'll fetch the data separately. (If you'd rather not poll at all, you can pass a webhook_url when queueing and receive the finished result via a signed POST.)

5. Page through the results

Results come back in pages. Follow next_cursor until it's null:

def iter_results(run_id, page_size=100):
    cursor = None
    while True:
        params = {"limit": page_size}
        if cursor:
            params["cursor"] = cursor
        r = requests.get(f"{BASE}/parser-runs/{run_id}/results",
                         params=params, headers=HEADERS, timeout=60)
        r.raise_for_status()
        body = r.json()
        yield from body["items"]
        cursor = body.get("next_cursor")
        if not cursor:
            break

6. Write a clean CSV

Overlapping searches (Austin vs. Round Rock) can return the same business twice, so dedupe on placeId. Also: treat every field as optional — plenty of businesses have no website or email.

FIELDS = ["title", "phone", "website", "emails", "address",
          "city", "rating", "reviewsCount", "category", "googleMapsUrl"]

seen = set()
with open("leads.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=FIELDS)
    w.writeheader()
    for item in iter_results(run_id):
        pid = item.get("placeId") or item.get("googleMapsUrl")
        if pid in seen:
            continue
        seen.add(pid)
        row = {k: item.get(k) for k in FIELDS}
        row["emails"] = "; ".join(item.get("emails") or [])
        w.writerow(row)

print(f"saved {len(seen)} unique businesses to leads.csv")

That's it — run the script and you get a leads.csv you can open in Excel/Sheets or import into a CRM.

Things worth remembering (for any scraping API)

Bound every job. A limit on results is your first line of cost control; a max budget is the second.
Idempotency keys are not optional once you add retries. Without one, "the request timed out" can silently mean "you now have two jobs".
Check the job status, not just the HTTP status. Async APIs often report failures inside a 200 response.
Store the run ID before you start polling, so a crash doesn't lose a job you already paid for.
Data is messy. Phones aren't verified mobiles, emails aren't verified deliverable, and fields can be missing. Validate before you import.
Scrape responsibly — only collect data you're allowed to process and follow applicable laws and the source's terms.

If you want to try this without writing code first, the same scraper runs from a form in the browser and exports JSON/CSV/XLSX; new accounts get 2,500 free credits a month. Full API docs: extracto.cloud/docs.

What would you build on top of a lead list like this — enrichment, a local-SEO audit, a price-comparison map? Let me know in the comments.

DEV Community: Extracto