python 75 lines · 8 steps

A concurrent batch geocoder in asyncio

A queue, a pool of workers, and a semaphore turn a list of place names into coordinates without hammering the API.

Explained by highlit
Advanced asyncio producer-consumer rate-limiting concurrency http-client
1import asyncio
2import logging
3from dataclasses import dataclass
4 
5import httpx
6 
7logger = logging.getLogger("geocoder")
8 
9NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
10 
11 
12@dataclass
13class GeocodeResult:
14 query: str
15 lat: float | None
16 lon: float | None
17 display_name: str | None = None
18 
19 
20async def _geocode_one(client: httpx.AsyncClient, query: str) -> GeocodeResult:
21 resp = await client.get(
22 NOMINATIM_URL,
23 params={"q": query, "format": "jsonv2", "limit": 1},
24 headers={"User-Agent": "batch-geocoder/1.0"},
25 )
26 resp.raise_for_status()
27 hits = resp.json()
28 if not hits:
29 return GeocodeResult(query, None, None)
30 top = hits[0]
31 return GeocodeResult(query, float(top["lat"]), float(top["lon"]), top["display_name"])
32 
33 
34async def _worker(
35 name: int,
36 queue: asyncio.Queue[str],
37 client: httpx.AsyncClient,
38 limiter: asyncio.Semaphore,
39 results: list[GeocodeResult],
40) -> None:
41 while True:
42 query = await queue.get()
43 try:
44 async with limiter:
45 result = await _geocode_one(client, query)
46 await asyncio.sleep(1.0) # Nominatim: max 1 req/s
47 results.append(result)
48 except httpx.HTTPError as exc:
49 logger.warning("worker %d failed on %r: %s", name, query, exc)
50 results.append(GeocodeResult(query, None, None))
51 finally:
52 queue.task_done()
53 
54 
55async def batch_geocode(
56 queries: list[str], *, workers: int = 4, max_concurrency: int = 1
57) -> list[GeocodeResult]:
58 queue: asyncio.Queue[str] = asyncio.Queue()
59 for q in queries:
60 queue.put_nowait(q)
61 
62 results: list[GeocodeResult] = []
63 limiter = asyncio.Semaphore(max_concurrency)
64 
65 async with httpx.AsyncClient(timeout=15.0) as client:
66 tasks = [
67 asyncio.create_task(_worker(i, queue, client, limiter, results))
68 for i in range(workers)
69 ]
70 await queue.join()
71 for task in tasks:
72 task.cancel()
73 await asyncio.gather(*tasks, return_exceptions=True)
74 
75 return results
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A shared queue plus a fixed worker pool is the classic way to bound concurrency while draining a work list.
  2. 2A semaphore held across the request and its cooldown enforces a hard rate limit independent of worker count.
  3. 3queue.join() lets the coordinator wait for completion, then cancel idle workers cleanly.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
go 
func (w *Watcher) resetDebounce(d time.Duration) {
	if !w.timer.Stop() {
		select {
		case <-w.timer.C:

Debouncing a stream of events in Go

debounce timers channels
Advanced 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
rust 
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
 

Building a counting semaphore in Rust

concurrency synchronization condition-variable
Advanced 9 steps

Share this explainer

Here's the card — post it anywhere.

A concurrent batch geocoder in asyncio — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code