A useful CSV starts with a small, explainable search. Before importing business records into a spreadsheet, you need to know which query produced them, whether the data is simulated, and which fields were missing.
This tutorial turns a G Maps Crawl Search API response into a local CSV using Python's standard library. The script saves the original JSON, limits the number of pages, and remembers request identifiers so a timeout does not force you to start a new search.
The example searches restaurants in Ballard, Seattle. Substitute a category and location that fit your research. Results are a sample of returned listings; they do not establish complete coverage of a neighborhood.
Understand the request before writing code
Send a flat JSON body to POST https://gmapscrawl.com/api/v1/search:
{
"q": "restaurants in Ballard Seattle",
"page": 1,
"hl": "en",
"extra": false
}
Authentication uses Authorization: Bearer YOUR_API_KEY. The response returns businesses synchronously under data.businesses; there is no job-status endpoint to poll for this operation. Each business's listing fields sit under place.
A page returns at most 20 businesses. Page numbers run from 1 through 10, and a successful live page costs one API request unit, including an empty result. Deduplicating ten overlapping rows locally does not refund the search request. A two-page budget can return up to 40 rows before deduplication and use up to two units.
Select the API product in your workspace and use a key with scrapes:write. API allowances are separate from Online and Extension subscriptions. If your workspace offers a test key, start there: test responses contain data.simulated: true, use zero request units, and contain fixtures rather than real prospects. Check the key environment before calling the endpoint.
Prepare a private run directory
Use Python 3.10 or later. There are no third-party packages to install. Supply GMSCRAPER_API_KEY through your terminal's private environment or secret manager, rather than putting its value in the script or command arguments.
Save the following as maps_to_csv.py. Each output directory represents one fixed search run. Its manifest.json stores the query and a separate UUID for each page before a request is sent. It contains no API key.
The script deliberately makes no automatic retries. Successful page responses are reused from disk when you resume. A failed page keeps its original body and idempotency key for the next attempt. Run one process per directory; this small example is not a concurrent job runner.
import argparse
import csv
import json
import os
import sys
import uuid
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import HTTPRedirectHandler, Request, build_opener
ENDPOINT = "https://gmapscrawl.com/api/v1/search"
class NoRedirects(HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def save_json(path, value):
temporary = path.with_suffix(path.suffix + ".tmp")
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2),
encoding="utf-8")
temporary.replace(path)
def search(body, request_id, key):
request = Request(ENDPOINT, method="POST",
data=json.dumps(body).encode("utf-8"), headers={
"Authorization": "Bearer " + key,
"Content-Type": "application/json",
"Accept": "application/json",
"Idempotency-Key": request_id,
})
try:
with build_opener(NoRedirects()).open(request, timeout=180) as response:
return json.load(response)
except HTTPError as error:
retry_after = error.headers.get("Retry-After", "not provided")
try:
payload = json.loads(error.read())
code = payload.get("error", {}).get("code", "unknown")
trace = payload.get("request_id", "unknown")
except (ValueError, AttributeError):
code, trace = "non_json_error", "unknown"
raise RuntimeError(
f"HTTP {error.code}; code={code}; request_id={trace}; "
f"Retry-After={retry_after}. Stopped; keep this run directory."
) from None
except (URLError, TimeoutError) as error:
raise RuntimeError(
"Transport failed or timed out. Accepted work may still exist. "
"Resume this directory with unchanged arguments after review."
) from error
def spreadsheet_cell(value):
if value is None:
return ""
if not isinstance(value, str):
return value
# CSV quoting alone does not prevent spreadsheet formula evaluation.
if value.lstrip(" \t\r\n").startswith(("=", "+", "-", "@")) \
or value.startswith(("\t", "\r", "\n")):
return "'" + value
return value
FIELDS = ["place_id", "business_id", "name", "category", "address", "phone",
"website", "rating", "review_count", "maps_url", "observed_at",
"source_query", "source_page", "identity_review", "simulated",
"field_states"]
def write_csv(path, records):
temporary = path.with_suffix(".csv.tmp")
with temporary.open("w", encoding="utf-8-sig", newline="") as handle:
writer = csv.DictWriter(handle, fieldnames=FIELDS)
writer.writeheader()
for record, page, query, simulated in records.values():
place = record["place"]
row = dict(zip(FIELDS[:11], [
place.get("placeId"), record.get("businessId"),
place.get("name"), place.get("category"),
(place.get("address") or {}).get("formatted"),
place.get("phone"), place.get("website"),
place.get("rating"), place.get("reviewCount"),
place.get("mapsUrl"), place.get("observedAt"),
]))
row.update(source_query=query, source_page=page,
identity_review=not bool(place.get("placeId")),
simulated=simulated,
field_states=json.dumps(place.get("fieldStates", {}),
ensure_ascii=False))
writer.writerow({name: spreadsheet_cell(value)
for name, value in row.items()})
temporary.replace(path)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("query")
parser.add_argument("--pages", type=int, choices=range(1, 11), default=2)
parser.add_argument("--out", type=Path, required=True)
parser.add_argument("--allow-simulated", action="store_true")
args = parser.parse_args()
query = args.query.strip()
if not 2 <= len(query) <= 200:
raise RuntimeError("Query must contain 2–200 characters.")
config = {"q": query, "hl": "en", "extra": False, "max_pages": args.pages}
args.out.mkdir(parents=True, exist_ok=True)
manifest_path = args.out / "manifest.json"
manifest = (json.loads(manifest_path.read_text(encoding="utf-8"))
if manifest_path.exists() else {"config": config, "pages": {}})
if manifest["config"] != config:
raise RuntimeError("Arguments changed. Choose a NEW output directory.")
records = {}
simulation_mode = None
for page in range(1, args.pages + 1):
body = {"q": query, "page": page, "hl": "en", "extra": False}
entry = manifest["pages"].setdefault(str(page), {
"body": body, "id": str(uuid.uuid4()),
})
if entry["body"] != body:
raise RuntimeError("Saved request differs; do not reuse its key.")
save_json(manifest_path, manifest)
raw_path = args.out / f"page-{page:02}.json"
if raw_path.exists():
payload = json.loads(raw_path.read_text(encoding="utf-8"))
else:
key = os.environ.get("GMSCRAPER_API_KEY")
if not key:
raise RuntimeError("Set GMSCRAPER_API_KEY in your environment.")
payload = search(body, entry["id"], key)
save_json(raw_path, payload)
data = payload.get("data", {})
businesses = data.get("businesses")
if not isinstance(businesses, list) or len(businesses) > 20 \
or data.get("total") != len(businesses):
raise RuntimeError("Unexpected response shape; inspect saved JSON.")
simulated = data.get("simulated") is True
if simulation_mode is not None and simulation_mode != simulated:
raise RuntimeError("Mixed test/live pages. Start a separate run.")
simulation_mode = simulated
if simulated and not args.allow_simulated:
raise RuntimeError("Simulated fixture saved. Add --allow-simulated "
"only for a separate test-data run.")
for index, record in enumerate(businesses):
place = record.get("place")
if not isinstance(place, dict):
raise RuntimeError("Unexpected business record; inspect JSON.")
identity = (("place", place["placeId"]) if place.get("placeId")
else ("business", record["businessId"])
if record.get("businessId") else ("row", page, index))
records.setdefault(identity, (record, page, query, simulated))
write_csv(args.out / "businesses.csv", records)
if not businesses:
break
print(f"Saved {len(records)} unique records in {args.out}.")
if __name__ == "__main__":
try:
main()
except (RuntimeError, ValueError, OSError) as error:
print(f"Stopped: {error}", file=sys.stderr)
sys.exit(1)
Python's CSV module handles quoted commas, embedded newlines, and column delimiters. The HTTP request uses urllib.request, with redirects disabled so a credential-bearing request is not redirected to another destination.
Run a bounded search and inspect the files
For a test credential, use a clearly named fixture directory:
python3 maps_to_csv.py "restaurants in Ballard Seattle" \
--pages 1 --out ballard-fixture-01 --allow-simulated
For real research, configure a live key and use a new directory without the simulation flag:
python3 maps_to_csv.py "restaurants in Ballard Seattle" \
--pages 2 --out ballard-live-01
Inspect manifest.json, the saved page JSON files, and businesses.csv. Raw responses preserve the API envelope, missing values, and nested provenance. The CSV is a smaller working view. It carries source query, page, Maps URL, observation time, and simulation state alongside business details.
The script writes the CSV after every accepted page. If page two fails, page one's rows remain available, but the command exits unsuccessfully. Treat that file as partial output rather than a completed two-page search. An empty first page produces a header-only CSV and stops immediately.
Before using the file in a campaign or analysis, compare a few CSV rows with their corresponding saved JSON. Check a name containing a comma, a missing website, an international phone number, and a review count of zero. Confirm the spreadsheet has not converted an identifier into scientific notation. If the command stopped partway through, record which pages completed and resolve the error before describing the dataset as complete. These checks test the handoff you actually use, rather than assuming that a successfully created file imported correctly.
An output directory should stay private: responses can contain business contact information, and the manifest documents your research. Keep credentials and generated datasets out of a public repository. Share a reviewed extract when your team needs a sample.
Make retries deliberate
For a timeout or retryable upstream failure, rerun the identical command against the same directory after checking the error. The manifest preserves the failed page's UUID; successful saved pages need no network request. Accepted work can continue after a client timeout, so making up a new identifier risks starting a separate search.
For HTTP 429, wait at least the returned Retry-After period before retrying. Fix authentication, scope, validation, quota, or capability problems before attempting another call. A 409 idempotency conflict needs investigation of the original body, not a loop that keeps changing keys. The script reports errors and stops; it never sleeps and retries indefinitely. See the retry guide for the recovery rules.
Use a new run directory for a genuinely new query, a changed page budget, or a later acquisition you intentionally want to request. Resuming an old directory rebuilds its saved dataset; it does not refresh the source. Do not manually edit a stored body while keeping its request key.
Check identity, missing fields, and spreadsheet behavior
Rows with the same Place ID are deduplicated, keeping their first returned occurrence. A missing Place ID falls back to the service's businessId and sets identity_review to true. Names alone are never used to merge branches. Raw page files retain overlapping observations so you can review those decisions.
Blank CSV fields mean unknown or unavailable, not false. A missing website does not establish that the business has no site; an empty rating is different from a numerical zero. Read field_states and the original JSON before turning an absence into a sales qualification. Check the source location and category too: keyword searches can return nearby or related businesses.
The spreadsheet helper prefixes potentially formula-like text with an apostrophe, including phone strings beginning with +. This is a defensive transformation for the CSV; the untouched value remains in JSON. If your CRM expects the exact phone string, import it from JSON or use an explicit text-field mapping. Keep numerical ratings and review counts as numbers, and do not remove the CSV protection simply because quoting looks correct.
Add enrichment only after the basic workflow works
The script fixes extra to false. When your task needs available emails or social profiles, adapt a separate version to request extra: true, create a new run directory, and budget for its search pages. Enrichment checks business websites; it cannot guarantee an email for every listing or verify deliverability.
Current response time is not observation time. Query results normally have a 24-hour reuse window, with one hour for empty results. Complete Maps details may be reused for up to 730 days, while website contacts have a separate seven-day window. Preserve place.observedAt, place.contactsCheckedAt, and contact evidence rather than marking every row as freshly collected today. The business-field guide explains these distinctions.
Start with one or two pages, inspect a handful of rows, and verify that your destination imports identifiers and phone numbers correctly. Once the small run is explainable, you have a repeatable foundation for a larger integration. Prepare your API access.
