Google Maps API

Google Maps Platform API Integration

Implements production-grade Google Maps Platform integration using the googlemaps Python SDK and Google Maps REST API. When loaded, this skill makes the model implement geocoding (address → coordinates, coordinates → address), directions and route calculation with waypoints, places search (nearby, text, find place), distance matrix for multiple origins/destinations, time zone lookup by coordinates, elevation data, static maps, and street view imagery. All implementations follow Google Maps best practices: use GOOGLE_MAPS_API_KEY from environment, implement rate limiting and exponential backoff, cache geocoding results to reduce API calls, use session tokens for Place Autocomplete to control costs, handle API errors gracefully, and respect Google's Terms of Service regarding data storage and caching.

Pattern 2: Geocoding with Retry and Caching

import logging
import time
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class GeoResult:
    """Parsed geocoding result."""
    formatted_address: str
    latitude: float
    longitude: float
    place_id: str | None = None

    @property
    def location_str(self) -> str:
        return f"{self.latitude}, {self.longitude}"


class GeocodingClient:
    """Geocoding client with retry logic."""

    def __init__(self, api_key: str, max_retries: int = 3):
        self._api_key = api_key
        self._max_retries = max_retries

    def geocode(self, address: str) -> GeoResult:
        """Geocode an address string with retries."""
        last_error = None
        for attempt in range(1, self._max_retries + 1):
            try:
                response = self._call_geocoding_api(address)
                return self._parse_response(response)
            except Exception as e:
                last_error = e
                if attempt < self._max_retries:
                    time.sleep(2 ** attempt)
        raise ConnectionError(f"Geocoding failed after {self._max_retries} retries")

    def _call_geocoding_api(self, address: str) -> dict:
        """Call the Google Geocoding API."""
        # In production: requests.get(f"https://maps.googleapis.com/maps/api/geocode/json?address=...")
        return {"results": [{"formatted_address": "1600 Amphitheatre Pkwy",
                            "geometry": {"location": {"lat": 37.422, "lng": -122.085}}}],
                "status": "OK"}

    def _parse_response(self, response: dict) -> GeoResult:
        """Validate and parse API response."""
        if response.get("status") != "OK":
            raise ValueError(f"Geocoding API error: {response.get('status')}")
        result = response["results"][0]
        loc = result["geometry"]["location"]
        return GeoResult(
            formatted_address=result["formatted_address"],
            latitude=loc["lat"],
            longitude=loc["lng"],
        )

Constraints

MUST DO

• Implement structured error responses with consistent format: {error_code, message, details, request_id}
• Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm
• Validate all incoming requests against a schema before processing — reject malformed input with clear error messages
• Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries

MUST NOT DO

• Do not expose internal implementation details, stack traces, or database queries in error responses
• Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits
• Never trust client-supplied authentication tokens without validation (signature verification, expiration check)
• Do not log request/response bodies containing PII, API keys, or other sensitive data

TL;DR Checklist

• Use googlemaps Python SDK with API key from GOOGLE_MAPS_API_KEY env var
• Get API key from Google Cloud Console → Maps Platform → Credentials
• Enable required APIs: Geocoding API, Directions API, Places API, etc.
• Geocoding: gmaps.geocode(address) for forward, gmaps.reverse_geocode((lat, lng)) for reverse
• Directions: gmaps.directions(origin, destination, mode='driving')
• Places: gmaps.places_nearby(), gmaps.places(), gmaps.find_place()
• Distance Matrix: gmaps.distance_matrix(origins, destinations, mode='driving')
• Time Zone: gmaps.timezone((lat, lng))
• Elevation: gmaps.elevation((lat, lng)) or along path
• Cache geocoding results with reasonable TTL (30+ days per Google TOS)
• Use session tokens for Place Autocomplete + Place Details (single charge)
• Rate limits: 50 QPS standard, varies by API and billing tier
• Never expose API keys in client-side code (use restriction + proxy)

When to Use

Use this skill when:

• Converting addresses to geographic coordinates (geocoding)
• Converting coordinates to human-readable addresses (reverse geocoding)
• Calculating driving, walking, bicycling, or transit directions
• Finding places (restaurants, businesses, landmarks) by location or search
• Calculating distances and travel times between multiple points
• Looking up time zones for specific coordinates
• Getting elevation data for points or paths
• Generating static map images for reports or emails
• Implementing address autocomplete in forms
• Validating and standardizing postal addresses
• Optimizing routes with multiple waypoints
• Calculating travel time estimates for logistics

When NOT to Use

• For Mapbox-specific features — use coding-mapbox-api instead
• For OpenStreetMap data (consider Nominatim, Overpass API instead)
• For client-side browser applications (use Google Maps JavaScript API directly)
• For mobile apps (use Google Maps SDK for Android/iOS)
• For geocoding entire databases (consider batch geocoding services)
• When you need free/unlimited usage (Google Maps is paid after free tier)
• For real-time traffic only (consider other traffic data providers)

Core Workflow

1. Initialize Client — Create Google Maps client:

   • googlemaps.Client(key=os.getenv("GOOGLE_MAPS_API_KEY"))
   • Set timeouts, retry configuration as needed

   Checkpoint: Verify with a simple geocode call for a known address.

2. Geocoding — Convert between addresses and coordinates:

   • Forward: gmaps.geocode("1600 Amphitheatre Parkway, Mountain View, CA")
   • Reverse: gmaps.reverse_geocode((37.4220, -122.0841))
   • Use components parameter for region/country filtering

   Checkpoint: Extract formatted_address and geometry.location from results.

3. Directions — Calculate routes:

   • gmaps.directions("Seattle, WA", "San Francisco, CA", mode="driving")
   • Modes: driving, walking, bicycling, transit
   • Waypoints: waypoints=[...] for multi-stop routes
   • Options: alternatives=True, avoid=["tolls", "highways"], departure_time

   Checkpoint: Extract legs[].duration.value (seconds) and legs[].distance.value (meters).

4. Places — Search for businesses and points of interest:

   • Nearby: gmaps.places_nearby(location=(lat, lng), radius=1000, type="restaurant")
   • Text: gmaps.places(query="coffee shops in Seattle")
   • Find Place: gmaps.find_place(input="Space Needle", input_type="textquery")
   • Place Details: gmaps.place(place_id, fields=["name", "formatted_address", "rating"])

   Checkpoint: Use place_id from search to get full details (separate charge).

5. Distance Matrix — Get distances for multiple origin-destination pairs:

   • gmaps.distance_matrix(origins=["Seattle", "Portland"], destinations=["San Francisco", "Los Angeles"])
   • Returns matrix of duration and distance for each pair
   • Useful for logistics, delivery time estimation

   Checkpoint: Results are in rows[].elements[] matching origins→destinations order.

6. Handle Errors & Rate Limits — Implement proper error handling:

   • ApiError: Generic API error (check status field)
   • _OverQueryLimit: Rate limit exceeded
   • _ApiError with status="ZERO_RESULTS": No results found
   • Use exponential backoff for retryable errors

   Checkpoint: Rate limit handling includes jittered exponential backoff.

Implementation Patterns

Pattern 1: Google Maps Client Initialization (BAD vs GOOD)

"""Google Maps Platform client initialization patterns.

Key concepts:
- googlemaps: Official Google Maps Python SDK (github.com/googlemaps/google-maps-services-python)
- API Key: From Google Cloud Console → Maps Platform → Credentials
- Enable APIs: Geocoding API, Directions API, Places API, Distance Matrix API, etc.
- Rate Limits: Standard tier ~50 QPS, varies by API
- Pricing: Pay-as-you-go, $200 free credit/month
- Caching: Google allows caching geocoding results for up to 30 days (check TOS)

Environment variables:
    GOOGLE_MAPS_API_KEY: Your Google Maps API key
    GOOGLE_MAPS_QUERY_RATE: Optional queries per second rate limit
"""

from __future__ import annotations

import os
import json
import time
import logging
from typing import Any, Optional, List, Dict, Tuple, TypeVar, Callable, Generator
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from functools import wraps
from decimal import Decimal

logger = logging.getLogger(__name__)

# Try to import googlemaps
try:
    import googlemaps
    from googlemaps import exceptions as gmaps_exc
    GOOGLE_MAPS_AVAILABLE = True
except ImportError:
    GOOGLE_MAPS_AVAILABLE = False
    logger.warning("googlemaps not installed. Run: pip install googlemaps")


# ===================================================================
# ❌ BAD — hardcoded key, no error handling, no caching, client-side exposure
# ===================================================================

def bad_gmaps_init() -> Any:
    """❌ BAD: Don't do any of these things."""
    if not GOOGLE_MAPS_AVAILABLE:
        raise ImportError("googlemaps library required")
    
    # ❌ Hardcoded API key! Never commit this!
    # ❌ This would be visible in version control history
    gmaps = googlemaps.Client(key="AIzaSyDaGmWKa4JsXZ...")
    
    # ❌ No timeout configuration
    # ❌ No retry configuration
    # ❌ No error handling
    # ❌ No result caching (wastes API calls, money)
    # ❌ Using synchronous client in async context without consideration
    # ❌ Would expose key if used client-side (use HTTP referrer restrictions + proxy)
    
    return gmaps


# ===================================================================
# ✅ GOOD — env-based config, timeouts, retries, caching, error handling
# ===================================================================


class GoogleMapsError(Exception):
    """Base exception for Google Maps API errors."""
    
    def __init__(self, message: str, status: Optional[str] = None):
        super().__init__(message)
        self.status = status


class GoogleMapsAuthError(GoogleMapsError):
    """Authentication failed (invalid key, API not enabled, etc.)."""
    pass


class GoogleMapsRateLimitError(GoogleMapsError):
    """Rate limit exceeded (OVER_QUERY_LIMIT)."""
    
    def __init__(self, message: str, retry_after: Optional[float] = None):
        super().__init__(message)
        self.retry_after = retry_after


class GoogleMapsZeroResultsError(GoogleMapsError):
    """Query returned no results (ZERO_RESULTS)."""
    pass


class GoogleMapsInvalidRequestError(GoogleMapsError):
    """Invalid request parameters (INVALID_REQUEST)."""
    pass


# Map Google Maps status codes to our exceptions
GOOGLE_MAPS_STATUS_ERRORS = {
    "OVER_QUERY_LIMIT": GoogleMapsRateLimitError,
    "REQUEST_DENIED": GoogleMapsAuthError,
    "ZERO_RESULTS": GoogleMapsZeroResultsError,
    "INVALID_REQUEST": GoogleMapsInvalidRequestError,
    "UNKNOWN_ERROR": GoogleMapsError,
    "NOT_FOUND": GoogleMapsZeroResultsError,
    "MAX_ROUTE_LENGTH_EXCEEDED": GoogleMapsInvalidRequestError,
    "MAX_WAYPOINTS_EXCEEDED": GoogleMapsInvalidRequestError,
}


@dataclass
class GoogleMapsCacheEntry:
    """Cache entry for Google Maps API results."""
    
    result: Any
    cached_at: datetime
    ttl_seconds: int = 86400 * 30  # 30 days default (Google allows this for geocoding)
    
    def is_valid(self) -> bool:
        """Check if this cache entry is still valid."""
        now = datetime.now(timezone.utc)
        expires_at = self.cached_at + timedelta(seconds=self.ttl_seconds)
        return now < expires_at


@dataclass
class GoogleMapsConfig:
    """Google Maps configuration from environment variables.
    
    Environment variables:
        GOOGLE_MAPS_API_KEY: Required API key
        GOOGLE_MAPS_TIMEOUT: Optional timeout in seconds (default 30)
        GOOGLE_MAPS_RETRIES: Optional max retries (default 3)
        GOOGLE_MAPS_ENABLE_CACHE: Enable result caching (default true)
        GOOGLE_MAPS_CACHE_TTL_DAYS: Cache TTL in days (default 30)
    """
    
    # Required
    api_key: Optional[str] = None
    
    # Connection config
    timeout: float = 30.0
    connect_timeout: float = 10.0
    max_retries: int = 3
    queries_per_second: float = 10.0  # Conservative rate limit
    
    # Caching
    enable_cache: bool = True
    cache_ttl_days: int = 30  # Google allows 30 days for geocoding
    
    @classmethod
    def from_env(cls) -> "GoogleMapsConfig":
        """Load configuration from environment variables."""
        
        def parse_float(env_var: str, default: float) -> float:
            val = os.environ.get(env_var)
            if val is None:
                return default
            try:
                return float(val)
            except ValueError:
                return default
        
        def parse_int(env_var: str, default: int) -> int:
            val = os.environ.get(env_var)
            if val is None:
                return default
            try:
                return int(val)
            except ValueError:
                return default
        
        def parse_bool(env_var: str, default: bool) -> bool:
            val = os.environ.get(env_var)
            if val is None:
                return default
            return val.lower() in ("1", "true", "yes", "on")
        
        return cls(
            api_key=os.environ.get("GOOGLE_MAPS_API_KEY"),
            timeout=parse_float("GOOGLE_MAPS_TIMEOUT", 30.0),
            max_retries=parse_int("GOOGLE_MAPS_RETRIES", 3),
            queries_per_second=parse_float("GOOGLE_MAPS_QUERY_RATE", 10.0),
            enable_cache=parse_bool("GOOGLE_MAPS_ENABLE_CACHE", True),
            cache_ttl_days=parse_int("GOOGLE_MAPS_CACHE_TTL_DAYS", 30),
        )
    
    def is_enabled(self) -> bool:
        """Check if Google Maps is configured."""
        return bool(self.api_key)
    
    def validate(self) -> bool:
        """Validate configuration.
        
        Returns:
            True if valid
            
        Raises:
            ValueError: If invalid when enabled
        """
        if not self.is_enabled():
            logger.info("Google Maps not configured (GOOGLE_MAPS_API_KEY not set)")
            return True
        
        if not self.api_key or not self.api_key.strip():
            raise ValueError("GOOGLE_MAPS_API_KEY is empty")
        
        # Basic key format check (Google keys start with "AIza")
        if not self.api_key.startswith("AIza"):
            logger.warning(
                "GOOGLE_MAPS_API_KEY doesn't start with 'AIza' — "
                "Google Maps API keys typically start with 'AIza'"
            )
        
        return True


class GoogleMapsClient:
    """Production-grade Google Maps client with caching, retries, and rate limiting.
    
    Features:
    - Config from environment
    - Exponential backoff for rate limits
    - Configurable result caching (respects Google TOS)
    - Rate limiting (QPS control)
    - Unified error handling with meaningful exceptions
    - Convenience methods with result parsing
    """
    
    # Common travel modes
    MODE_DRIVING = "driving"
    MODE_WALKING = "walking"
    MODE_BICYCLING = "bicycling"
    MODE_TRANSIT = "transit"
    
    # Common place types (partial list)
    PLACE_ACCOUNTING = "accounting"
    PLACE_AIRPORT = "airport"
    PLACE_AMUSEMENT_PARK = "amusement_park"
    PLACE_AQUARIUM = "aquarium"
    PLACE_ART_GALLERY = "art_gallery"
    PLACE_ATM = "atm"
    PLACE_BAKERY = "bakery"
    PLACE_BANK = "bank"
    PLACE_BAR = "bar"
    PLACE_BEAUTY_SALON = "beauty_salon"
    PLACE_BICYCLE_STORE = "bicycle_store"
    PLACE_BOOK_STORE = "book_store"
    PLACE_BOWLING_ALLEY = "bowling_alley"
    PLACE_BUS_STATION = "bus_station"
    PLACE_CAFE = "cafe"
    PLACE_CAMPGROUND = "campground"
    PLACE_CAR_DEALER = "car_dealer"
    PLACE_CAR_RENTAL = "car_rental"
    PLACE_CAR_REPAIR = "car_repair"
    PLACE_CAR_WASH = "car_wash"
    PLACE_CASINO = "casino"
    PLACE_CEMETERY = "cemetery"
    PLACE_CHURCH = "church"
    PLACE_CITY_HALL = "city_hall"
    PLACE_CLOTHING_STORE = "clothing_store"
    PLACE_CONVENIENCE_STORE = "convenience_store"
    PLACE_COURTHOUSE = "courthouse"
    PLACE_DENTIST = "dentist"
    PLACE_DEPARTMENT_STORE = "department_store"
    PLACE_DOCTOR = "doctor"
    PLACE_ELECTRICIAN = "electrician"
    PLACE_ELECTRONICS_STORE = "electronics_store"
    PLACE_EMBASSY = "embassy"
    PLACE_FIRE_STATION = "fire_station"
    PLACE_FLORIST = "florist"
    PLACE_FUNERAL_HOME = "funeral_home"
    PLACE_FURNITURE_STORE = "furniture_store"
    PLACE_GAS_STATION = "gas_station"
    PLACE_GYM = "gym"
    PLACE_HAIR_CARE = "hair_care"
    PLACE_HARDWARE_STORE = "hardware_store"
    PLACE_HINDU_TEMPLE = "hindu_temple"
    PLACE_HOME_GOODS_STORE = "home_goods_store"
    PLACE_HOSPITAL = "hospital"
    PLACE_INSURANCE_AGENCY = "insurance_agency"
    PLACE_JEWELRY_STORE = "jewelry_store"
    PLACE_LAUNDRY = "laundry"
    PLACE_LAWYER = "lawyer"
    PLACE_LIBRARY = "library"
    PLACE_LIQUOR_STORE = "liquor_store"
    PLACE_LOCAL_GOVERNMENT_OFFICE = "local_government_office"
    PLACE_LOCKSMITH = "locksmith"
    PLACE_LODGING = "lodging"
    PLACE_MEAL_DELIVERY = "meal_delivery"
    PLACE_MEAL_TAKEAWAY = "meal_takeaway"
    PLACE_MOSQUE = "mosque"
    PLACE_MOVIE_RENTAL = "movie_rental"
    PLACE_MOVIE_THEATER = "movie_theater"
    PLACE_MOVING_COMPANY = "moving_company"
    PLACE_MUSEUM = "museum"
    PLACE_NIGHT_CLUB = "night_club"
    PLACE_PAINTER = "painter"
    PLACE_PARK = "park"
    PLACE_PARKING = "parking"
    PLACE_PET_STORE = "pet_store"
    PLACE_PHARMACY = "pharmacy"
    PLACE_PHYSIOTHERAPIST = "physiotherapist"
    PLACE_PLUMBER = "plumber"
    PLACE_POLICE = "police"
    PLACE_POST_OFFICE = "post_office"
    PLACE_REAL_ESTATE_AGENCY = "real_estate_agency"
    PLACE_RESTAURANT = "restaurant"
    PLACE_ROOFING_CONTRACTOR = "roofing_contractor"
    PLACE_RV_PARK = "rv_park"
    PLACE_SCHOOL = "school"
    PLACE_SHOE_STORE = "shoe_store"
    PLACE_SHOPPING_MALL = "shopping_mall"
    PLACE_SPA = "spa"
    PLACE_STADIUM = "stadium"
    PLACE_STORAGE = "storage"
    PLACE_STORE = "store"
    PLACE_SUBWAY_STATION = "subway_station"
    PLACE_SUPERMARKET = "supermarket"
    PLACE_SYNAGOGUE = "synagogue"
    PLACE_TAXI_STAND = "taxi_stand"
    PLACE_TRAIN_STATION = "train_station"
    PLACE_TRAVEL_AGENCY = "travel_agency"
    PLACE_UNIVERSITY = "university"
    PLACE_VETERINARY_CARE = "veterinary_care"
    PLACE_ZOO = "zoo"
    
    # Avoid parameters
    AVOID_TOLLS = "tolls"
    AVOID_HIGHWAYS = "highways"
    AVOID_FERRIES = "ferries"
    AVOID_INDOOR = "indoor"
    
    # Transit modes
    TRANSIT_MODE_BUS = "bus"
    TRANSIT_MODE_SUBWAY = "subway"
    TRANSIT_MODE_TRAIN = "train"
    TRANSIT_MODE_TRAM = "tram"
    TRANSIT_MODE_RAIL = "rail"
    
    # Traffic models
    TRAFFIC_MODEL_BEST_GUESS = "best_guess"
    TRAFFIC_MODEL_OPTIMISTIC = "optimistic"
    TRAFFIC_MODEL_PESSIMISTIC = "pessimistic"
    
    def __init__(self, config: GoogleMapsConfig):
        self._config = config
        self._client: Optional[googlemaps.Client] = None
        self._cache: Dict[str, GoogleMapsCacheEntry] = {}  # Simple in-memory cache
        self._last_request_time: float = 0.0
        self._min_request_interval: float = 1.0 / config.queries_per_second
    
    def _get_client(self) -> googlemaps.Client:
        """Get or create the Google Maps client.
        
        Lazy-loaded.
        """
        if not GOOGLE_MAPS_AVAILABLE:
            raise ImportError(
                "googlemaps not installed. Run: pip install googlemaps"
            )
        
        self._config.validate()
        
        if self._client is None:
            self._client = googlemaps.Client(
                key=self._config.api_key,
                timeout=self._config.timeout,
                connect_timeout=self._config.connect_timeout,
                retry_over_query_limit=True,  # Basic retry in SDK
            )
            logger.info("Google Maps client initialized")
        
        return self._client
    
    def _rate_limit_wait(self) -> None:
        """Wait if needed to respect QPS rate limit."""
        import time
        
        now = time.time()
        elapsed = now - self._last_request_time
        
        if elapsed < self._min_request_interval:
            wait_time = self._min_request_interval - elapsed
            logger.debug("Rate limiting: waiting %.3fs", wait_time)
            time.sleep(wait_time)
        
        self._last_request_time = time.time()
    
    def _get_cache_key(self, method: str, args: tuple, kwargs: dict) -> str:
        """Generate a cache key for an API call.
        
        Simple but effective: combine method name with normalized args.
        """
        import hashlib
        
        # Normalize args for consistent caching
        key_parts = [method]
        
        for arg in args:
            if isinstance(arg, (list, tuple)):
                # Normalize coordinate tuples
                if len(arg) == 2 and all(isinstance(x, (int, float, str)) for x in arg):
                    # Coordinates: normalize to (lat, lng) strings with 6 decimals
                    try:
                        lat, lng = float(arg[0]), float(arg[1])
                        key_parts.append(f"({lat:.6f},{lng:.6f})")
                        continue
                    except (ValueError, TypeError):
                        pass
            key_parts.append(str(arg))
        
        # Include relevant kwargs
        cacheable_kwargs = {}
        for key, value in kwargs.items():
            if key in ("mode", "language", "region", "components", "avoid", 
                      "units", "departure_time", "arrival_time",
                      "traffic_model", "transit_mode", "transit_routing_preference",
                      "radius", "type", "keyword", "rankby"):
                cacheable_kwargs[key] = value
        
        if cacheable_kwargs:
            key_parts.append(str(sorted(cacheable_kwargs.items())))
        
        key_string = "|".join(key_parts)
        return hashlib.md5(key_string.encode()).hexdigest()
    
    def _get_from_cache(self, cache_key: str) -> Optional[Any]:
        """Get a result from cache if available and valid."""
        if not self._config.enable_cache:
            return None
        
        entry = self._cache.get(cache_key)
        
        if entry and entry.is_valid():
            logger.debug("Cache hit for key: %s...", cache_key[:16])
            return entry.result
        
        if entry:
            # Expired, remove
            del self._cache[cache_key]
        
        return None
    
    def _store_in_cache(self, cache_key: str, result: Any) -> None:
        """Store a result in cache."""
        if not self._config.enable_cache:
            return
        
        entry = GoogleMapsCacheEntry(
            result=result,
            cached_at=datetime.now(timezone.utc),
            ttl_seconds=self._config.cache_ttl_days * 86400,
        )
        
        self._cache[cache_key] = entry
        logger.debug("Cached result for key: %s...", cache_key[:16])
    
    def _execute_with_retry(
        self,
        operation: Callable[[], Any],
        method_name: str,
        cache_key: Optional[str] = None,
        cacheable: bool = True,
    ) -> Any:
        """Execute a Google Maps API operation with caching, retries, and rate limits.
        
        Args:
            operation: Callable that makes the API call
            method_name: Name for logging
            cache_key: Cache key (if None, no caching for this call)
            cacheable: Whether this result can be cached
            
        Returns:
            API response (parsed)
            
        Raises:
            GoogleMapsError: Various error types based on response
        """
        import random
        
        # Check cache first if applicable
        if cacheable and cache_key:
            cached = self._get_from_cache(cache_key)
            if cached is not None:
                return cached
        
        # Rate limit before making request
        self._rate_limit_wait()
        
        delay = 1.0
        last_exception: Optional[Exception] = None
        
        for attempt in range(self._config.max_retries):
            try:
                # Get client each time (handles lazy init)
                client = self._get_client()
                
                # Execute the operation
                result = operation()
                
                # The googlemaps SDK returns parsed JSON directly
                # It raises exceptions for error statuses
                
                # Cache if cacheable
                if cacheable and cache_key:
                    self._store_in_cache(cache_key, result)
                
                return result
                
            except gmaps_exc._OverQueryLimit as e:
                last_exception = e
                
                wait_time = min(
                    delay * (2 ** attempt) + random.uniform(0, 0.5),
                    60.0
                )
                
                logger.warning(
                    "Google Maps rate limited on %s (attempt %d/%d). Waiting %.1fs",
                    method_name, attempt + 1, self._config.max_retries, wait_time
                )
                
                time.sleep(wait_time)
                continue
            
            except gmaps_exc.ApiError as e:
                # Extract status if available
                status = getattr(e, "status", None)
                message = str(e)
                
                # Map to our exception types
                error_class = GOOGLE_MAPS_STATUS_ERRORS.get(status, GoogleMapsError)
                
                # Some cases are retryable
                if status == "UNKNOWN_ERROR" and attempt < self._config.max_retries - 1:
                    last_exception = e
                    wait_time = min(delay * (2 ** attempt), 30.0)
                    logger.warning(
                        "Google Maps UNKNOWN_ERROR on %s (attempt %d/%d). Retrying in %.1fs",
                        method_name, attempt + 1, self._config.max_retries, wait_time
                    )
                    time.sleep(wait_time)
                    continue
                
                raise error_class(message, status=status) from e
            
            except Exception as e:
                # Non-API exceptions (network, timeout, etc.)
                last_exception = e
                
                if attempt < self._config.max_retries - 1:
                    wait_time = min(delay * (2 ** attempt), 30.0)
                    logger.warning(
                        "Request error on %s (attempt %d/%d): %s. Retrying in %.1fs",
                        method_name, attempt + 1, self._config.max_retries, e, wait_time
                    )
                    time.sleep(wait_time)
                    continue
                
                raise GoogleMapsError(f"Google Maps operation failed: {e}") from e
        
        # All retries exhausted
        raise GoogleMapsRateLimitError(
            f"Rate limit retries exhausted after {self._config.max_retries} attempts"
        ) from last_exception
    
    # ===================================================================
    # Geocoding
    # ===================================================================
    
    def geocode(
        self,
        address: str,
        components: Optional[Dict[str, str]] = None,
        bounds: Optional[Tuple[Tuple[float, float], Tuple[float, float]]] = None,
        region: Optional[str] = None,
        language: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Geocode an address to coordinates.
        
        Converts a human-readable address to geographic coordinates
        (latitude/longitude) and complete address components.
        
        Args:
            address: Address to geocode (e.g., "1600 Amphitheatre Parkway, Mountain View, CA")
            components: Component filtering (e.g., {"country": "US", "locality": "Seattle"})
            bounds: Optional viewport bias ((southwest_lat, southwest_lng), (northeast_lat, northeast_lng))
            region: Region code (e.g., "us", "uk") for biasing
            language: Language code for results (e.g., "en", "es", "fr")
            
        Returns:
            List of geocoding results (usually 1 for specific addresses)
            Each result has: formatted_address, geometry.location (lat/lng),
            address_components, place_id, types, etc.
            
        Raises:
            GoogleMapsZeroResultsError: If no results found
        """
        kwargs: Dict[str, Any] = {}
        if components:
            kwargs["components"] = components
        if bounds:
            kwargs["bounds"] = bounds
        if region:
            kwargs["region"] = region
        if language:
            kwargs["language"] = language
        
        def _op():
            return self._get_client().geocode(address, **kwargs)
        
        cache_key = self._get_cache_key("geocode", (address,), kwargs)
        
        return self._execute_with_retry(_op, f"geocode({address[:50]})", cache_key=cache_key)
    
    def geocode_first(
        self,
        address: str,
        components: Optional[Dict[str, str]] = None,
        **kwargs: Any,
    ) -> Optional[Dict[str, Any]]:
        """Geocode and return only the first result, or None.
        
        Convenience method for when you expect exactly one match.
        
        Args:
            address: Address to geocode
            components: Component filtering
            **kwargs: Additional geocode parameters
            
        Returns:
            First result dict, or None if no results
        """
        try:
            results = self.geocode(address, components=components, **kwargs)
            return results[0] if results else None
        except GoogleMapsZeroResultsError:
            return None
    
    def geocode_to_coords(
        self,
        address: str,
        components: Optional[Dict[str, str]] = None,
        **kwargs: Any,
    ) -> Optional[Tuple[float, float]]:
        """Geocode address and return only (latitude, longitude) tuple.
        
        Args:
            address: Address to geocode
            components: Component filtering
            **kwargs: Additional parameters
            
        Returns:
            (lat, lng) tuple, or None if not found
        """
        result = self.geocode_first(address, components=components, **kwargs)
        
        if not result:
            return None
        
        geometry = result.get("geometry", {})
        location = geometry.get("location", {})
        
        lat = location.get("lat")
        lng = location.get("lng")
        
        if lat is not None and lng is not None:
            return (float(lat), float(lng))
        
        return None
    
    def reverse_geocode(
        self,
        location: Tuple[float, float],
        result_type: Optional[List[str]] = None,
        location_type: Optional[List[str]] = None,
        language: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Reverse geocode coordinates to address.
        
        Converts latitude/longitude to human-readable addresses.
        
        Args:
            location: (latitude, longitude) tuple
            result_type: Filter by result types (e.g., ["street_address", "premise"])
            location_type: Filter by location type (["rooftop", "approximate"])
            language: Language code for results
            
        Returns:
            List of address results (most relevant first)
        """
        kwargs: Dict[str, Any] = {}
        if result_type:
            kwargs["result_type"] = result_type
        if location_type:
            kwargs["location_type"] = location_type
        if language:
            kwargs["language"] = language
        
        def _op():
            return self._get_client().reverse_geocode(location, **kwargs)
        
        cache_key = self._get_cache_key("reverse_geocode", (location,), kwargs)
        
        return self._execute_with_retry(_op, f"reverse_geocode({location})", cache_key=cache_key)
    
    def reverse_geocode_first(
        self,
        location: Tuple[float, float],
        **kwargs: Any,
    ) -> Optional[Dict[str, Any]]:
        """Reverse geocode and return first result.
        
        Args:
            location: (lat, lng) tuple
            **kwargs: Additional parameters
            
        Returns:
            First result, or None
        """
        try:
            results = self.reverse_geocode(location, **kwargs)
            return results[0] if results else None
        except GoogleMapsZeroResultsError:
            return None
    
    def reverse_geocode_to_address(
        self,
        location: Tuple[float, float],
        **kwargs: Any,
    ) -> Optional[str]:
        """Reverse geocode and return formatted address string.
        
        Args:
            location: (lat, lng) tuple
            **kwargs: Additional parameters
            
        Returns:
            Formatted address string, or None
        """
        result = self.reverse_geocode_first(location, **kwargs)
        
        if result:
            return result.get("formatted_address")
        
        return None
    
    # ===================================================================
    # Address Components Extraction
    # ===================================================================
    
    @staticmethod
    def extract_address_components(
        geocode_result: Dict[str, Any],
    ) -> Dict[str, Any]:
        """Extract structured address components from a geocode result.
        
        Converts the address_components list into a key-value dict
        with common fields like street_number, route, locality, etc.
        
        Args:
            geocode_result: Single result from geocode() or reverse_geocode()
            
        Returns:
            Dict with keys like: street_number, route, locality (city),
            administrative_area_level_1 (state), postal_code, country, etc.
        """
        components = geocode_result.get("address_components", [])
        result: Dict[str, Any] = {}
        
        # Map Google's types to common field names
        type_mapping = {
            "street_number": "street_number",
            "route": "street",
            "locality": "city",
            "administrative_area_level_1": "state",
            "administrative_area_level_2": "county",
            "postal_code": "postal_code",
            "postal_code_suffix": "postal_code_suffix",
            "country": "country",
            "sublocality": "sublocality",
            "neighborhood": "neighborhood",
            "premise": "premise",
            "subpremise": "subpremise",
            "floor": "floor",
            "room": "room",
        }
        
        for comp in components:
            types = comp.get("types", [])
            long_name = comp.get("long_name")
            short_name = comp.get("short_name")
            
            for google_type, field_name in type_mapping.items():
                if google_type in types:
                    result[field_name] = long_name
                    result[f"{field_name}_short"] = short_name
                    break
        
        # Also add formatted and geometry
        result["formatted_address"] = geocode_result.get("formatted_address")
        result["place_id"] = geocode_result.get("place_id")
        
        geometry = geocode_result.get("geometry", {})
        location = geometry.get("location", {})
        if location:
            result["latitude"] = location.get("lat")
            result["longitude"] = location.get("lng")
        
        # Also extract viewport if present
        viewport = geometry.get("viewport", {})
        if viewport:
            northeast = viewport.get("northeast", {})
            southwest = viewport.get("southwest", {})
            result["viewport"] = {
                "northeast": {"lat": northeast.get("lat"), "lng": northeast.get("lng")},
                "southwest": {"lat": southwest.get("lat"), "lng": southwest.get("lng")},
            }
        
        return result
    
    # ===================================================================
    # Directions
    # ===================================================================
    
    def directions(
        self,
        origin: str | Tuple[float, float],
        destination: str | Tuple[float, float],
        mode: str = "driving",
        waypoints: Optional[List[str | Tuple[float, float]]] = None,
        alternatives: bool = False,
        avoid: Optional[List[str]] = None,
        language: Optional[str] = None,
        units: Optional[str] = None,
        region: Optional[str] = None,
        departure_time: Optional[int | datetime] = None,
        arrival_time: Optional[int | datetime] = None,
        optimize_waypoints: bool = False,
        traffic_model: Optional[str] = None,
        transit_mode: Optional[List[str]] = None,
        transit_routing_preference: Optional[str] = None,
    ) -> List[Dict[str, Any]]:
        """Get directions from origin to destination.
        
        Calculates directions and distance/duration for routes.
        
        Args:
            origin: Start address or (lat, lng) tuple
            destination: End address or (lat, lng) tuple
            mode: Travel mode: 'driving', 'walking', 'bicycling', 'transit'
            waypoints: List of intermediate waypoints
            alternatives: Return multiple alternative routes
            avoid: Features to avoid: ['tolls', 'highways', 'ferries', 'indoor']
            language: Language code for instructions
            units: 'metric' or 'imperial'
            region: Region code for biasing
            departure_time: Departure time (for traffic-based duration)
            arrival_time: Arrival time (for transit)
            optimize_waypoints: Let Google optimize waypoint order
            traffic_model: Traffic model: 'best_guess', 'optimistic', 'pessimistic'
            transit_mode: Transit modes: ['bus', 'subway', 'train', 'tram', 'rail']
            transit_routing_preference: 'less_walking' or 'fewer_transfers'
            
        Returns:
            List of route dicts with legs, steps, distance, duration
        """
        kwargs: Dict[str, Any] = {}
        if waypoints:
            kwargs["waypoints"] = waypoints
        if alternatives:
            kwargs["alternatives"] = alternatives
        if avoid:
            kwargs["avoid"] = avoid
        if language:
            kwargs["language"] = language
        if units:
           

…(truncated)