Configuring Google Maps Fallback to OpenStreetMap

Configure a deterministic fallback chain that attempts the Google Maps Geocoding API first, intercepts explicit failure states, and re-routes the identical address payload to OpenStreetMap Nominatim — all as part of the Implementing Fallback Chains for Failed Lookups pattern.

Provider API Parameter Reference

The two providers share no common request schema. The table below captures every parameter that the fallback wrapper must set correctly for each provider before sending a request.

Failure-State Routing Table

Not every Google Maps failure warrants a Nominatim call. Routing to the secondary provider for client-side errors wastes quota and adds latency. The decision must be driven by the specific status code returned.

Fallback Request Flow

The diagram below shows the decision path a single address takes through the routing wrapper — from the initial Google call through failure-state evaluation to the Nominatim fallback and final schema normalization.

Minimal Runnable Implementation

The class below compiles to a single-import, drop-in component. It handles schema normalization, timeout management, Nominatim rate-limit compliance, and structured logging in under 100 lines.

import time
import logging
from typing import Dict, Optional

import requests
from requests.exceptions import RequestException, Timeout, HTTPError

logger = logging.getLogger(__name__)

# Provider endpoints — defined at module level for clarity
_GOOGLE_URL: str = "https://maps.googleapis.com/maps/api/geocode/json"
_OSM_URL: str = "https://nominatim.openstreetmap.org/search"
_OSM_MIN_INTERVAL: float = 1.0  # Nominatim Usage Policy: max 1 req/sec


class GeoFallbackResolver:
    """
    Routes geocoding requests to Google Maps, then Nominatim on failure.

    Handles schema normalization, timeout management, and strict
    Nominatim rate-limit compliance (1 req/sec per the usage policy).
    """

    def __init__(
        self,
        google_api_key: str,
        user_agent: str = "DataPipeline/1.0 ([email protected])",
    ) -> None:
        self.google_api_key = google_api_key
        self._osm_headers: Dict[str, str] = {"User-Agent": user_agent}
        self._last_osm_request: float = 0.0

    # ------------------------------------------------------------------
    # Rate limiting
    # ------------------------------------------------------------------

    def _enforce_osm_rate_limit(self) -> None:
        """Block until Nominatim's 1 req/sec policy is satisfied."""
        elapsed = time.time() - self._last_osm_request
        if elapsed < _OSM_MIN_INTERVAL:
            time.sleep(_OSM_MIN_INTERVAL - elapsed)
        self._last_osm_request = time.time()

    # ------------------------------------------------------------------
    # Schema normalizers
    # ------------------------------------------------------------------

    def _normalize_google(self, payload: Dict) -> Optional[Dict]:
        if payload.get("status") != "OK" or not payload.get("results"):
            return None
        loc = payload["results"][0]["geometry"]["location"]
        return {
            "lat": loc["lat"],
            "lon": loc["lng"],  # Google uses "lng", not "lon"
            "address": payload["results"][0].get("formatted_address", ""),
            "provider": "google",
        }

    def _normalize_osm(self, payload: list) -> Optional[Dict]:
        if not payload:
            return None
        result = payload[0]
        return {
            "lat": float(result["lat"]),
            "lon": float(result["lon"]),
            "address": result.get("display_name", ""),
            "provider": "osm",
        }

    # ------------------------------------------------------------------
    # Public resolver
    # ------------------------------------------------------------------

    def resolve(self, address: str, timeout: float = 5.0) -> Dict:
        """
        Attempt Google Maps; fall back to Nominatim on failure states.

        Returns a dict with keys: lat, lon, address, provider.
        provider is "none" when both resolvers fail.
        """
        # 1. Primary: Google Maps Geocoding
        try:
            params = {"address": address, "key": self.google_api_key}
            resp = requests.get(_GOOGLE_URL, params=params, timeout=timeout)
            resp.raise_for_status()
            data = resp.json()

            google_status = data.get("status", "")
            if google_status == "INVALID_REQUEST":
                logger.error("Malformed address, halting chain: %s", address)
                return {"lat": None, "lon": None, "address": address, "provider": "none"}

            if google_status in ("OVER_QUERY_LIMIT", "REQUEST_DENIED"):
                logger.warning(
                    "Google quota/auth failure (%s), falling back to OSM",
                    google_status,
                )
            else:
                normalized = self._normalize_google(data)
                if normalized:
                    return normalized
                logger.warning(
                    "Google returned status=%s for: %s", google_status, address
                )
        except (Timeout, HTTPError, RequestException) as exc:
            logger.warning("Google network failure for '%s': %s", address, exc)

        # 2. Fallback: OpenStreetMap Nominatim
        logger.info("Delegating to OSM for: %s", address)
        try:
            self._enforce_osm_rate_limit()
            params = {"q": address, "format": "json", "limit": 1}
            resp = requests.get(
                _OSM_URL, params=params, headers=self._osm_headers, timeout=timeout
            )
            resp.raise_for_status()
            normalized = self._normalize_osm(resp.json())
            if normalized:
                return normalized
            logger.warning("OSM returned empty results for: %s", address)
        except (Timeout, HTTPError, RequestException) as exc:
            logger.error("OSM fallback failed for '%s': %s", address, exc)

        return {"lat": None, "lon": None, "address": address, "provider": "none"}

Vectorized pandas usage

import pandas as pd

resolver = GeoFallbackResolver(google_api_key="YOUR_KEY")

# resolve() is synchronous; use a progress-bar-friendly apply
df = pd.DataFrame({"raw_address": ["1600 Pennsylvania Ave NW, Washington DC", "10 Downing St, London"]})
results = df["raw_address"].apply(resolver.resolve).apply(pd.Series)
df = pd.concat([df, results], axis=1)
# df now has columns: raw_address, lat, lon, address, provider

For high-volume batches, wrap the resolver in a thread-pool executor (concurrent.futures.ThreadPoolExecutor) or convert _enforce_osm_rate_limit to await asyncio.sleep and switch to an async HTTP client. The Building Async Geocoding Requests in Python guide covers the async conversion in detail.

Edge Cases and Failure Modes

1. ZERO_RESULTS on a Valid Address

Google returns ZERO_RESULTS for addresses that exist but fall outside its coverage — typically rural routes, informal settlements, or recently constructed streets. The fallback to Nominatim is warranted here: Nominatim’s OSM data often includes community-mapped addresses that commercial providers miss. However, if the address is structurally garbled (missing city/state, transposed house number), Nominatim will also return an empty list. Flag these records for upstream address parsing and standardization rather than re-querying indefinitely.

# Adjusted: always try OSM on ZERO_RESULTS
if google_status in ("OVER_QUERY_LIMIT", "REQUEST_DENIED", "ZERO_RESULTS"):
    # fall through to Nominatim block — no early return
    pass

2. Nominatim Serving a Geocoding Artifact

Nominatim occasionally returns a match at the centroid of an administrative boundary rather than a precise street address — this happens when only the city or postal code is indexed for a given record. The display_name will be unusually short (no street component) and the coordinates will be at the centre of a city or district. Validate by checking that display_name contains at least three comma-separated components before accepting the result.

def _normalize_osm(self, payload: list) -> Optional[Dict]:
    if not payload:
        return None
    result = payload[0]
    display_name: str = result.get("display_name", "")
    if display_name.count(",") < 2:
        # Centroid match — too coarse for address-level use
        return None
    return {
        "lat": float(result["lat"]),
        "lon": float(result["lon"]),
        "address": display_name,
        "provider": "osm",
    }

3. Sustained OVER_QUERY_LIMIT Without a Circuit Breaker

If Google quota is exhausted for a multi-hour batch run, every request will hit Nominatim. Without a circuit breaker, the mandatory 1-second inter-request delay throttles the entire pipeline to 1 address/second. Implement a simple counter-based circuit breaker that bypasses the Google call entirely once consecutive OVER_QUERY_LIMIT responses exceed a threshold, and reopens only after a cooldown period.

class GeoFallbackResolver:
    _OPEN_THRESHOLD = 5      # failures before opening the circuit
    _COOLDOWN_SECONDS = 90   # seconds before probing Google again

    def __init__(self, google_api_key: str, user_agent: str = "DataPipeline/1.0") -> None:
        self.google_api_key = google_api_key
        self._osm_headers: Dict[str, str] = {"User-Agent": user_agent}
        self._last_osm_request: float = 0.0
        self._google_fail_count: int = 0
        self._circuit_open_until: float = 0.0

    def _google_circuit_open(self) -> bool:
        return time.time() < self._circuit_open_until

    def _record_google_failure(self) -> None:
        self._google_fail_count += 1
        if self._google_fail_count >= self._OPEN_THRESHOLD:
            self._circuit_open_until = time.time() + self._COOLDOWN_SECONDS
            logger.warning("Google circuit opened for %d s", self._COOLDOWN_SECONDS)
            self._google_fail_count = 0

Integration Note

This page covers the concrete two-provider configuration. It slots directly into the broader Implementing Fallback Chains for Failed Lookups pattern, which defines the full provider priority matrix, exponential backoff strategy, and async executor. Once the resolver is wired up, route its provider tag into your API quota tracking and cost management layer so that Nominatim fallback events appear as a separate cost line — a sustained fallback rate above 20–30% indicates a quota misconfiguration or budget issue that needs addressing at the infrastructure level, not in the resolver itself.

---

Related

• Implementing Fallback Chains for Failed Lookups — the parent guide covering provider priority matrices, stateful request context, and async chain execution.
• API Quota Tracking and Cost Management — track per-provider spend and set budget-based circuit breakers so Nominatim fallback events are surfaced as a cost signal.
• Building Async Geocoding Requests in Python — convert the synchronous resolver above to a non-blocking asyncio/httpx implementation for high-throughput batch pipelines.
• Core Address Parsing & Standardization — normalize and validate address strings before sending them to any geocoding provider to reduce ZERO_RESULTS and INVALID_REQUEST failures.