Python Geocoding API Tutorial: Async, Retries, and a 100K-Row CSV Pipeline

This is a working Python tutorial for geocoding addresses through a REST API. Every snippet in this post was written, copy-pasted into a file, and executed against the live API on a clean Python 3.11 install before publishing. Two dependencies (httpx, aiofiles) — no SDK, no framework, no magic.

By the end of the post you will have a script that streams a CSV of any size, geocodes its rows concurrently with a configurable rate limit, retries on 429s with exponential backoff that respects the server's Retry-After header, and writes a {lat, lng, error} CSV without ever loading the input file into memory. About 50 lines of Python.

The endpoint used throughout is csv2geo.com/api/v1. The free tier is 1,000 forward/reverse requests per day plus 100 batch rows per day, no credit card required. Sign in and grab a key from /api-keys if you want to follow along — the demo endpoint used in a couple of early curl examples needs no key.

The endpoint

Two endpoints cover 95% of real-world geocoding work. Forward turns an address string into coordinates. Reverse turns coordinates into an address. Both accept GET (single) or POST (batch).

# Forward (single)
GET https://csv2geo.com/api/v1/geocode?q=ADDRESS&country=US

# Reverse (single)
GET https://csv2geo.com/api/v1/reverse?lat=LAT&lng=LNG

# Batch forward
POST https://csv2geo.com/api/v1/geocode
Body: { "addresses": ["addr1", "addr2", ...] }

# Auth: either ?api_key=KEY query string, or
# Authorization: Bearer KEY header

The response shape for a single forward request looks like this. This is real output, not a documentation example.

{
  "query": "1600 Pennsylvania Ave NW Washington DC",
  "results": [
    {
      "formatted_address": "1600 Pennsylvania Ave NW, Washington, DC 20500-0005, United States",
      "location": { "lat": 38.89768, "lng": -77.03655 },
      "accuracy": "houseNumber",
      "accuracy_score": 1,
      "components": {
        "house_number": "1600",
        "street": "Pennsylvania Ave NW",
        "city": "Washington",
        "state": "District of Columbia",
        "postal_code": "20500-0005",
        "country": "USA"
      }
    }
  ],
  "meta": { "response_time_ms": 673, "source": "here" }
}

Two fields matter most when writing client code. results[0]["location"] is your {lat, lng}. results[0]["accuracy"] is the match level — "houseNumber" is rooftop, "street" is street centroid, "place" is a POI match, "postcode" is a postcode centroid. Use it to drop low-confidence rows before they pollute your dataset. The numeric accuracy_score (0.0 – 1.0) gives you a finer threshold; see geocoding confidence scores explained for how to pick one.

First request

Standard library urllib works, but httpx is the right tool — it has the same API for sync and async, modern timeouts, and proper connection pooling.

pip install httpx

# geocode.py
import os
import httpx

API_KEY = os.environ["CSV2GEO_KEY"]
BASE = "https://csv2geo.com/api/v1"

def geocode(address: str, country: str = "US") -> dict | None:
    r = httpx.get(
        f"{BASE}/geocode",
        params={"q": address, "country": country},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10.0,
    )
    r.raise_for_status()
    data = r.json()
    results = data.get("results") or []
    return results[0]["location"] if results else None

if __name__ == "__main__":
    print(geocode("1 Apple Park Way, Cupertino, CA"))
    # -> {'lat': 37.33177, 'lng': -122.03042}

Run it:

CSV2GEO_KEY=geo_live_xxx python geocode.py

The Authorization: Bearer header form is the recommended one. Keys never end up in your shell history or your access logs that way, unlike the ?api_key=... query-string form. Also note r.raise_for_status() — httpx does not raise on non-2xx by default, so you have to ask for it.

Reverse geocoding

Same shape, different parameters. Pass lat and lng, get back an address.

def reverse(lat: float, lng: float) -> dict | None:
    r = httpx.get(
        f"{BASE}/reverse",
        params={"lat": lat, "lng": lng},
        headers={"Authorization": f"Bearer {API_KEY}"},
        timeout=10.0,
    )
    r.raise_for_status()
    results = r.json().get("results") or []
    return results[0] if results else None

# Eiffel Tower
hit = reverse(48.8584, 2.2945)
print(hit["formatted_address"], hit.get("distance_meters"))

Reverse responses include a distance_meters field — how far the matched address sits from the input coordinate. If you are reverse-geocoding GPS pings from a delivery app, anything over ~50m usually means the GPS fix was bad, not the geocoder. Below ~10m and you are looking at the rooftop. The ground-truth distance is the only honest accuracy metric for reverse geocoding; everything else lies a little.

Doing it in parallel

The wrong way to geocode 10,000 addresses in Python is asyncio.gather over a list comprehension of fetches. That fires 10,000 concurrent connections at the API, hits the rate limit on the second batch, and exhausts your local file descriptors at the same time. Don't.

The right way is bounded concurrency. The standard idiom in Python is an asyncio.Semaphore, plus a single shared httpx.AsyncClient so connections are reused.

# async_geocode.py
import asyncio
import os
import httpx

API_KEY = os.environ["CSV2GEO_KEY"]
BASE = "https://csv2geo.com/api/v1"
CONCURRENCY = 8  # tune to your plan's per-minute rate limit

async def geocode_one(client: httpx.AsyncClient, sem: asyncio.Semaphore,
                      address: str, country: str = "US") -> dict | None:
    async with sem:
        r = await client.get(
            "/geocode",
            params={"q": address, "country": country},
        )
        r.raise_for_status()
        results = r.json().get("results") or []
        return results[0]["location"] if results else None

async def geocode_many(addresses: list[str]) -> list[dict | None]:
    sem = asyncio.Semaphore(CONCURRENCY)
    headers = {"Authorization": f"Bearer {API_KEY}"}
    limits = httpx.Limits(max_connections=CONCURRENCY, max_keepalive_connections=CONCURRENCY)
    async with httpx.AsyncClient(base_url=BASE, headers=headers,
                                 timeout=15.0, limits=limits) as client:
        return await asyncio.gather(*(
            geocode_one(client, sem, a) for a in addresses
        ))

if __name__ == "__main__":
    addrs = ["1600 Pennsylvania Ave NW Washington DC",
             "1 Apple Park Way, Cupertino, CA",
             "233 S Wacker Dr, Chicago, IL"]
    print(asyncio.run(geocode_many(addrs)))

Concurrency is empirical, but a useful starting rule is: take your plan's per-minute rate limit, divide by 60, and aim for roughly that many in-flight requests. Free is 100/min — start at 4. Starter ($49, 1K/min) — 16. Growth ($149, 5K/min) — 64. Pro ($499, 10K/min) — 128 and watch the headers. The full breakdown is in concurrency tuning for geocoding.

Two design choices in the code worth pointing out. First, the asyncio.Semaphore is acquired *inside* the per-task coroutine, not outside the gather call. That way the coroutines are all scheduled immediately but only N actually run network I/O at once. Second, httpx.Limits mirrors the semaphore — without it, httpx will happily open more sockets than your semaphore permits if your server is slower than expected.

Retry, backoff, and rate-limit headers

Three things every production geocoder needs: (1) treat 429 and 5xx as retriable, (2) honour the Retry-After header when the server sends one, (3) cap the number of attempts so a permanently dead key does not loop forever.

import asyncio
import random
import httpx

class GeocodeError(Exception):
    pass

async def geocode_with_retry(client: httpx.AsyncClient, address: str,
                             country: str = "US", *, max_attempts: int = 5) -> dict:
    for attempt in range(1, max_attempts + 1):
        try:
            r = await client.get("/geocode",
                                 params={"q": address, "country": country})
        except (httpx.TimeoutException, httpx.TransportError) as e:
            if attempt == max_attempts:
                raise GeocodeError(f"network error after {attempt} attempts: {e}") from e
            await asyncio.sleep(_backoff(attempt))
            continue

        if r.status_code < 300:
            return r.json()

        # Non-retriable: 4xx other than 429 means bad key, bad input, etc.
        if r.status_code != 429 and r.status_code < 500:
            raise GeocodeError(f"HTTP {r.status_code}: {r.text}")

        # Retriable: 429 / 5xx. Honour Retry-After if present.
        if attempt == max_attempts:
            raise GeocodeError(f"gave up after {attempt} attempts (last={r.status_code})")
        retry_after = r.headers.get("Retry-After")
        delay = float(retry_after) if retry_after else _backoff(attempt)
        await asyncio.sleep(delay)

    raise GeocodeError("unreachable")

def _backoff(attempt: int) -> float:
    # 1, 2, 4, 8, 16... seconds + jitter
    return (2 ** (attempt - 1)) + random.uniform(0, 1)

A few notes. The Retry-After header may be an integer number of seconds or an HTTP-date; in practice the API always sends seconds, so float(...) is safe. Network errors (timeouts, connection resets) are retried just like 5xx — they are usually transient. The jitter in _backoff matters when many workers retry simultaneously: without it, every worker wakes up at the same instant and re-DDoSes the server. See exponential backoff: when to retry, when to stop for the math on retry budgets and dead-letter queues.

The response also exposes three headers worth checking on every successful call:

| Header | Meaning | |---|---| | X-RateLimit-Limit | Your plan's per-minute ceiling | | X-RateLimit-Remaining | What you have left in this window | | X-RateLimit-Reset | Unix seconds until the window resets | | Retry-After | Sent on 429s — seconds to wait before retrying |

If Remaining drops below 10% of Limit, slow down voluntarily — sleep, lower your semaphore, switch to batch. Cheaper than thrashing on 429s and far cheaper than a sudden burst that takes the whole pipeline offline. The deeper theory of which limiter algorithm produces these headers is in token bucket vs leaky bucket vs sliding window.

The batch endpoint

Subscription tiers from Starter ($49/mo) upward expose a batch endpoint that takes an array of addresses in a single POST. Plan caps:

| Plan | Monthly rows | Per-minute | Batch size | |---|---|---|---| | Free | — (1K/day API, 100/day batch) | 100 | 100 | | Starter ($49) | 50,000 | 1,000 | 1,000 | | Growth ($149) | 250,000 | 5,000 | 5,000 | | Pro ($499) | 1,000,000 | 10,000 | 10,000 |

async def batch_geocode(client: httpx.AsyncClient,
                        addresses: list[str]) -> list[dict]:
    r = await client.post("/geocode", json={"addresses": addresses})
    r.raise_for_status()
    return r.json()["results"]

# Usage:
async def main():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with httpx.AsyncClient(base_url=BASE, headers=headers,
                                 timeout=60.0) as client:
        out = await batch_geocode(client, [
            "1600 Pennsylvania Ave NW, Washington, DC",
            "1 Apple Park Way, Cupertino, CA",
            "233 S Wacker Dr, Chicago, IL",
        ])
        for i, hit in enumerate(out):
            loc = hit["results"][0]["location"] if hit.get("results") else None
            print(i, loc)

Batch responses preserve input order: position N of the response always corresponds to position N of the input. No need to round-trip an id field, though you can. Note the higher client timeout — a 1,000-address batch on Growth takes about 10–15 seconds end-to-end.

Streaming a CSV without OOM

If your input file is 100,000 rows, you do not want to pandas.read_csv it, hold the DataFrame in RAM, and then apply over it. You will run out of memory or burn through your daily quota in one minute. Stream it.

pip install httpx aiofiles

# stream_geocode.py
import asyncio
import csv
import os
import sys
import httpx
import aiofiles

API_KEY = os.environ["CSV2GEO_KEY"]
BASE = "https://csv2geo.com/api/v1"
CONCURRENCY = 8
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

async def geocode_row(client: httpx.AsyncClient, sem: asyncio.Semaphore,
                      row: dict) -> dict:
    async with sem:
        try:
            r = await client.get("/geocode", params={"q": row["address"]})
        except httpx.HTTPError as e:
            return {**row, "lat": "", "lng": "", "error": f"network:{type(e).__name__}"}
        if r.status_code != 200:
            return {**row, "lat": "", "lng": "", "error": f"http_{r.status_code}"}
        results = r.json().get("results") or []
        if not results:
            return {**row, "lat": "", "lng": "", "error": "no_match"}
        top = results[0]
        if top.get("accuracy_score", 0) < 0.7:
            return {**row, "lat": "", "lng": "", "error": "low_confidence"}
        loc = top["location"]
        return {**row, "lat": loc["lat"], "lng": loc["lng"], "error": ""}

async def main(in_path: str, out_path: str) -> None:
    sem = asyncio.Semaphore(CONCURRENCY)
    limits = httpx.Limits(max_connections=CONCURRENCY)
    async with httpx.AsyncClient(base_url=BASE, headers=HEADERS,
                                 timeout=15.0, limits=limits) as client, \
               aiofiles.open(out_path, "w", newline="") as out_f:
        # Write header first
        await out_f.write("id,address,lat,lng,error\n")

        async def handle(row: dict):
            res = await geocode_row(client, sem, row)
            line = f'{res["id"]},"{res["address"]}",{res["lat"]},{res["lng"]},{res["error"]}\n'
            await out_f.write(line)

        with open(in_path, newline="") as in_f:
            reader = csv.DictReader(in_f)
            tasks = [asyncio.create_task(handle(row)) for row in reader]
            await asyncio.gather(*tasks)

    print(f"done -> {out_path}")

if __name__ == "__main__":
    asyncio.run(main(sys.argv[1], sys.argv[2]))

Three things this script does that beginners' code usually does not:

• Reads the input as a stream (no full-file load) and writes the output line-by-line (no buffer-everything-then-flush).
• Caps in-flight geocoding work at CONCURRENCY concurrent requests via a semaphore — the input loop schedules tasks immediately, but only 8 actually hit the network at once.
• Records errors in a column rather than crashing on the first failure. That is the difference between a script that finishes and a script that has to be restarted with a "where did it stop" question.

For inputs over a few hundred thousand rows, you also want to bound the *task list itself*, not just network concurrency — otherwise the unawaited tasks consume RAM. Wrap the reader in batches of, say, 5,000 and asyncio.gather each batch before reading the next chunk. The same pattern, with one extra outer loop.

A minimal Pydantic model

If you would rather not pass dict everywhere, type the response shape once and let Pydantic validate it on parse.

pip install pydantic

# models.py
from typing import Literal
from pydantic import BaseModel, Field

Accuracy = Literal["houseNumber", "street", "place", "postcode"]

class Location(BaseModel):
    lat: float
    lng: float

class Components(BaseModel):
    house_number: str | None = None
    street: str | None = None
    city: str | None = None
    state: str | None = None
    postal_code: str | None = None
    country: str | None = None

class GeocodeResult(BaseModel):
    formatted_address: str
    location: Location
    accuracy: Accuracy
    accuracy_score: float = Field(ge=0.0, le=1.0)
    components: Components

class Meta(BaseModel):
    response_time_ms: int
    source: str

class GeocodeResponse(BaseModel):
    query: str
    results: list[GeocodeResult]
    meta: Meta

# Usage
def geocode_typed(address: str) -> GeocodeResult | None:
    r = httpx.get(f"{BASE}/geocode",
                  params={"q": address},
                  headers={"Authorization": f"Bearer {API_KEY}"},
                  timeout=10.0)
    r.raise_for_status()
    parsed = GeocodeResponse.model_validate(r.json())
    return parsed.results[0] if parsed.results else None

Two design choices worth pointing out. First, accuracy is a Literal union, so a match over it without a wildcard is a type error — your IDE will flag it the day the API adds a fifth value. Second, accuracy_score has a Field(ge=0.0, le=1.0) constraint, so a malformed upstream response fails parse rather than poisoning a downstream filter. That is the kind of typed code that survives years of API minor versions.

Things that bite people

httpx does not raise on non-2xx by default. A 404, 429, or 503 produces a normal Response object whose .status_code you have to check. If you skip raise_for_status() and trust the body, you will eventually parse {"error": "..."} as if it were a successful result. Either always call raise_for_status() or always check r.status_code explicitly.

Country codes matter — and `country` is a hint, not a filter. If you pass country="US" for an address in Toronto, you will get back a wrong but plausible result somewhere in upstate New York. ISO alpha-2 is the right format (DE for Germany, GB for the UK, BR for Brazil). When your input is mixed, set country per row from a column rather than passing one default for the whole file.

Empty strings are valid input that returns garbage. Validate row["address"] for whitespace and minimum length before sending. Two minutes of input sanitation saves an hour of "why is the lat 0,0?" investigation. The full pre-geocoding cleanup checklist is in address parsing before geocoding.

Calling async code from a sync context. asyncio.run(coro) works once per process. Inside Jupyter (which already has a running event loop) or inside an existing FastAPI request, it raises RuntimeError: asyncio.run() cannot be called from a running event loop. Inside Jupyter use await coro directly; inside FastAPI make your handler async def and await the geocode function.

Frequently Asked Questions

Do I need an SDK?

No. The API is REST + JSON, and httpx covers sync and async with one library. The "SDK" most teams end up writing is the geocode_with_retry function above plus a wrapper for the batch endpoint and a Pydantic model. About 60 lines. Maintaining your own tiny wrapper is cheaper than tracking an SDK release cadence.

How do I get types?

The Pydantic model in the section above is the minimal viable typing. If you do not want a runtime dependency, the same shapes work as TypedDict in typing — narrow accuracy to a Literal["houseNumber", "street", "place", "postcode"] and you have IDE autocomplete on every result field.

How do I integrate this with Django or Flask?

Both frameworks are sync-first by default, so the simplest path is the sync httpx.get form inside a view function. For a Flask endpoint that geocodes a single address, the geocode() function from the "First request" section drops in unchanged. For a Django view, the same. If you need fan-out (one request, many internal geocodes), use httpx.AsyncClient inside an async def view (Django 4.1+ supports this natively) or a Celery task that runs asyncio.run(geocode_many(...)).

How do I tune concurrency?

Take your plan's per-minute rate limit, divide by 60, and aim for roughly that many concurrent requests. 1,000/min ÷ 60 ≈ 17, so a CONCURRENCY of 16 is a safe default on Starter. The truth source is the X-RateLimit-Remaining header — log it on every successful response and watch for it dropping faster than 1/req. If it does, you have over-allocated; halve concurrency. Rule of thumb: optimal concurrency is a curve, and the elbow is usually near (rate_limit / 60) * 1.5.

How do I detect a no-match versus a successful low-confidence match?

results is an empty list on no-match. On a low-confidence match, results[0] exists but accuracy is "postcode" or "place" rather than "houseNumber" or "street", and accuracy_score is well below 1.0. A reasonable default threshold is 0.7: anything below, treat as no-match. Stricter pipelines (insurance risk scoring, healthcare patient mapping) use 0.95 and require accuracy == "houseNumber". The full picture is in geocoding confidence scores explained.

Does this work for non-US addresses?

Yes. Pass the right ISO alpha-2 in the country parameter — DE for Germany, GB for the UK, BR for Brazil, JP for Japan. Coverage spans 39 countries today, including the full top 10 by address count: USA (121M addresses), Brazil (90M), Mexico (30M), France (26M), Italy (26M), and the rest. The API page lists per-country counts.

Should I use single requests or the batch endpoint?

Batch when you have ≥100 addresses to do at once and your plan allows it (Starter+). One POST is cheaper for both sides than 100 GETs, and the latency saving is roughly avg_per_request_ms * count / concurrency. Singles are simpler for streaming workloads where addresses arrive over time, and singles are the only option on the free tier (100 batch rows/day cap aside). The full tradeoff analysis is in batch vs realtime geocoding.

Where to go from here

The full reference for the API is at csv2geo.com/api. If you would rather work in Node, the Node.js geocoding tutorial follows the same structure with fetch and p-limit. For deeper dives on the patterns this post touches: rate limiting algorithms compared, exponential backoff: when to retry, when to stop, caching geocoding results for 90% cost reduction, and geocoding confidence scores explained.

If you find the X-RateLimit headers undercounting in some edge case, or you hit a response shape this post does not cover, the contact form on the site reaches a person who reads it. Bug reports with curl (or httpx) reproductions get fixed quickly.

I.A. / CSV2GEO Creator

Related Articles

• Node.js Geocoding API Tutorial: Concurrency, Retries, and CSV Streaming
• Rate Limiting a Geocoding Pipeline: Token Bucket vs Leaky Bucket vs Sliding Window
• Exponential Backoff for Geocoding APIs: When to Retry, When to Stop
• How to Cache Geocoding Results: TTL, Keys, and 90% Cost Reduction
• Geocoding Confidence Scores Explained: When to Trust, When to Reject