Bulk geocoding
Geocode a whole CSV. Keep every result.
Run a spreadsheet of addresses through the search API at your plan's speed, or send millions of rows to a dedicated Enterprise cluster. Store the coordinates with no deletion deadline, subject to the data licences.
How long will it take?
Every plan has a per-second limit and a monthly allowance; there is no daily cap. Each successful search spends one credit, including one that finds nothing. Errors and rejected calls are free.
| Plan | Requests / second | Addresses / hour | Monthly allowance | Allowance at full speed |
|---|---|---|---|---|
| Free $0/month | 2 | 7,200 | 50,000 | ~7 h |
| Starter $29/month | 5 | 18,000 | 300,000 | ~17 h |
| Growth $99/month | 25 | 90,000 | 1,500,000 | ~17 h |
| Enterprise | Dedicated capacity sized to your deadline, with a flat price. 1,700+/s forward searches from a single server pod in our benchmark. | |||
Hours and per-hour figures are the plan's per-second limit sustained without pauses; network time and retries add to them.
Batch geocode a CSV
One script, standard library only.
It reads the address column, calls /v1/search in parallel
without exceeding your plan's per-second limit, and writes coordinates, the matched address
and goodMatch next to your original columns.
It waits out rate limits and retries temporary errors, and stops when your credits run out, keeping the rows written so far.
Filter on good_match = False afterwards to review partial matches
and addresses that weren't found.
export OPENGEOCODING_API_KEY=YOUR_API_KEY
export REQUESTS_PER_SECOND=5
python3 geocode_csv.py customers.csv customers-geocoded.csv"""Geocode the "address" column of a CSV: python3 geocode_csv.py in.csv out.csv"""
import csv
import json
import os
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.error import HTTPError
from urllib.request import Request, urlopen
API_URL = os.environ.get('OPENGEOCODING_API_URL', 'https://api.opengeocoding.com') + '/v1/search'
API_KEY = os.environ['OPENGEOCODING_API_KEY']
REQUESTS_PER_SECOND = int(os.environ.get('REQUESTS_PER_SECOND', '5')) # your plan's limit
def search(address):
for attempt in range(6):
request = Request(
API_URL,
data=json.dumps({'addressString': address}).encode(),
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
method='POST',
)
try:
with urlopen(request, timeout=30) as response:
return json.load(response)
except HTTPError as error:
retry_after = error.headers.get('Retry-After')
if error.code == 429 and retry_after:
time.sleep(float(retry_after))
elif error.code == 503:
time.sleep(2**attempt)
else:
# 429 without Retry-After means the credits ran out.
raise SystemExit(f'{error.code} for {address!r}: {error.read().decode()}')
raise SystemExit(f'Gave up on {address!r}')
def geocode(row):
# Each worker spends at least one second per row, so N workers stay under N requests/second.
started = time.monotonic()
result = search(row['address']) if row['address'].strip() else {}
time.sleep(max(0, 1 - (time.monotonic() - started)))
point = result.get('geoCoordinates', {})
return {
**row,
'lat': point.get('lat', ''),
'lon': point.get('lon', ''),
'matched_address': result.get('displayName', ''),
'good_match': result.get('goodMatch', False),
}
with open(sys.argv[1], newline='', encoding='utf-8') as source:
reader = csv.DictReader(source)
rows = list(reader)
columns = [*reader.fieldnames, 'lat', 'lon', 'matched_address', 'good_match']
with open(sys.argv[2], 'w', newline='', encoding='utf-8') as target:
writer = csv.DictWriter(target, columns)
writer.writeheader()
with ThreadPoolExecutor(REQUESTS_PER_SECOND) as pool:
for done, row in enumerate(pool.map(geocode, rows), 1):
writer.writerow(row)
print(f'\r{done}/{len(rows)}', end='', file=sys.stderr)
print(file=sys.stderr)
Geocoding a database, not a spreadsheet?
Enterprise runs your file on dedicated cores and hands it back geocoded, with flat pricing, an SLA and the option to run the engine in your own cloud.