Python Forward Geocode for Radar

radar.py

import httpx
import sentry_sdk
from pydantic import BaseModel
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)


class Geometry(BaseModel):
    type: str
    coordinates: list[float]


class TimeZone(BaseModel):
    id: str
    name: str
    code: str
    currentTime: str
    utcOffset: int
    dstOffset: int


class Address(BaseModel):
    latitude: float
    longitude: float
    geometry: Geometry
    country: str
    countryCode: str
    countryFlag: str
    county: str | None = None
    city: str | None = None
    state: str | None = None
    stateCode: str | None = None
    postalCode: str | None = None
    layer: str
    formattedAddress: str
    addressLabel: str
    timeZone: TimeZone | None = None
    distance: float | None = None
    confidence: str | None = None
    borough: str | None = None
    neighborhood: str | None = None
    number: str | None = None
    street: str | None = None


class Meta(BaseModel):
    code: int


class GeocodeResponse(BaseModel):
    meta: Meta
    addresses: list[Address]


class RadarClient:
    """
    A client class for interacting with the Radar.io Geocoding API using httpx.
    This class handles authentication via API key and provides methods for forward geocoding
    with exponential backoff on errors using the tenacity library.
    """

    def __init__(self, api_key: str):
        """
        Initializes the RadarClient with the provided API key.

        :param api_key: The publishable API key for authentication.
        """
        if not api_key:
            raise ValueError("API key must be provided.")
        self.api_key: str = api_key
        self.base_url: str = "https://api.radar.io/v1/geocode/forward"

    @retry(
        stop=stop_after_attempt(6),  # Initial attempt + 5 retries
        wait=wait_exponential(
            multiplier=1, min=0, max=32
        ),  # Backoff: 1, 2, 4, 8, 16 seconds, capped at 32
        retry=retry_if_exception_type(httpx.HTTPError),  # Retry on HTTP errors
        reraise=True,  # Reraise the last exception after retries
    )
    def _make_request(self, params: dict[str, str]) -> dict[str, object]:
        """
        Internal method to make the HTTP GET request.

        :param params: Dictionary of query parameters.
        :return: The JSON response as a dictionary.
        """
        headers: dict[str, str] = {"Authorization": self.api_key}
        response: httpx.Response = httpx.get(
            self.base_url, params=params, headers=headers
        )
        response.raise_for_status()
        return response.json()

    def forward_geocode(
        self,
        query: str,
        layers: str | None = None,
        country: str | None = None,
        lang: str | None = None,
    ) -> GeocodeResponse:
        """
        Performs forward geocoding to convert an address to coordinates.

        :param query: The address or place name to geocode (required).
        :param layers: Optional comma-separated layer filters (e.g., 'address,locality').
        :param country: Optional comma-separated 2-letter country codes (e.g., 'US,CA').
        :param lang: Optional language code for results (e.g., 'en', defaults to 'en').
        :return: The parsed GeocodeResponse object.
        :raises ValueError: If the query is not provided.
        :raises httpx.HTTPError: If the request fails after retries.
        :raises pydantic.ValidationError: If the response does not match the expected structure.
        """
        if not query:
            raise ValueError("Query parameter is required.")

        params: dict[str, str] = {"query": query}
        if layers:
            params["layers"] = layers
        if country:
            params["country"] = country
        if lang:
            params["lang"] = lang

        raw_response = self._make_request(params)
        return GeocodeResponse.model_validate(raw_response)


class GeocodeResult(BaseModel):
    """Result of geocoding a location with extracted coordinates and address info."""

    lat: float
    lon: float
    zip_code: str

    # in some cases, Radar or a geoprovider may not have this
    city: str | None
    state: str | None


def geocode_zip_code(
    radar_client: RadarClient, zip_code: str, country: str = "US"
) -> GeocodeResult:
    """
    Geocode a zip code and extract coordinates and address information.

    Handles error cases with Sentry logging and returns a standardized result.

    Args:
        radar_client: The radar client to use for geocoding
        zip_code: The zip code to geocode
        country: Country code (default: "US")

    Returns:
        GeocodeResult with lat, lon, city, and state information.
        Returns (0.0, 0.0, None, None) if geocoding fails.
    """
    location_result = radar_client.forward_geocode(zip_code, country=country)

    lat, lon = 0.0, 0.0
    city = None
    state = None

    if len(location_result.addresses) == 0:
        sentry_sdk.capture_message(
            "No geocoding results for zip code",
            level="info",
            extras={"zip": zip_code, "country": country},
        )
    else:
        if len(location_result.addresses) > 1:
            sentry_sdk.capture_message(
                "Multiple geocoding results for zip code",
                extras={
                    "zip": zip_code,
                    "results": len(location_result.addresses),
                },
            )
        address = location_result.addresses[0]
        lat = address.geometry.coordinates[1]
        lon = address.geometry.coordinates[0]
        state = address.stateCode
        city = address.city

    return GeocodeResult(lat=lat, lon=lon, city=city, state=state, zip_code=zip_code)