g2s-aggregator/app/adapters/address_suggestions/tomtom/client.py

"""TomTom address suggestion adapter."""

from typing import Any
from urllib.parse import quote

import httpx

from app.adapters.address_suggestions.base import (
    AddressSuggestionClientError,
    AddressSuggestionProvider,
    AddressSuggestionRequestError,
)
from app.config import TomTomAddressSuggestionsConfig
from app.schemas.request import AddressSuggestRequest
from app.schemas.response import AddressSuggestion

_TOMTOM_IDX_SET = "PAD,Addr,Str,EPP"


class TomTomClientError(AddressSuggestionClientError):
    """Raised when TomTom Search requests fail."""


class TomTomRequestError(TomTomClientError, AddressSuggestionRequestError):
    """Raised when TomTom rejects request parameters."""


class TomTomAddressSuggestionProvider(AddressSuggestionProvider):
    name = "tomtom"

    def __init__(
        self,
        http_client: httpx.AsyncClient,
        *,
        url: str,
        api_key: str,
        timeout_seconds: float = 10.0,
    ) -> None:
        self._http_client = http_client
        self._url = url.rstrip("/")
        self._api_key = api_key
        self._timeout_seconds = timeout_seconds

    @classmethod
    def from_config(
        cls,
        *,
        http_client: httpx.AsyncClient,
        config: TomTomAddressSuggestionsConfig,
    ) -> "TomTomAddressSuggestionProvider":
        return cls(
            http_client=http_client,
            url=config.url,
            api_key=config.api_key,
            timeout_seconds=config.timeout_seconds,
        )

    async def suggest(self, request: AddressSuggestRequest) -> list[AddressSuggestion]:
        request_url = self._build_request_url(request)
        params = self._build_query_params(request)
        try:
            response = await self._http_client.get(
                request_url,
                params=params,
                timeout=self._timeout_seconds,
            )
        except (httpx.TimeoutException, httpx.TransportError) as exc:
            raise TomTomClientError("TomTom Search request failed.") from exc

        if response.status_code == 400:
            raise TomTomRequestError(
                "TomTom Search request was rejected with status 400."
            )

        if response.status_code >= 500 or response.status_code in {403, 429}:
            raise TomTomClientError(
                "TomTom Search request failed with status "
                f"{response.status_code}."
            )

        if 400 <= response.status_code < 500:
            raise TomTomClientError(
                "TomTom Search request failed with status "
                f"{response.status_code}."
            )

        try:
            response.raise_for_status()
            raw_payload = response.json()
        except (httpx.HTTPError, TypeError, ValueError) as exc:
            raise TomTomClientError("TomTom Search returned invalid payload.") from exc

        try:
            return _map_tomtom_response(raw_payload)
        except ValueError as exc:
            raise TomTomClientError(
                "TomTom Search response payload is invalid."
            ) from exc

    def _build_request_url(self, request: AddressSuggestRequest) -> str:
        search_query = f"{request.city.strip()} {request.query.strip()}".strip()
        return f"{self._url}/{quote(search_query, safe='')}.json"

    def _build_query_params(self, request: AddressSuggestRequest) -> dict[str, Any]:
        params: dict[str, Any] = {
            "key": self._api_key,
            "countrySet": request.country_code.strip().upper(),
            "typeahead": "true",
            "idxSet": _TOMTOM_IDX_SET,
        }
        if request.limit is not None:
            params["limit"] = request.limit
        return params


def _map_tomtom_response(payload: object) -> list[AddressSuggestion]:
    if not isinstance(payload, dict):
        raise ValueError("TomTom response must be a JSON object.")

    raw_results = payload.get("results")
    if raw_results is None:
        return []
    if not isinstance(raw_results, list):
        raise ValueError("TomTom response must include results list.")

    return [_map_tomtom_result(item) for item in raw_results]


def _map_tomtom_result(payload: object) -> AddressSuggestion:
    if not isinstance(payload, dict):
        raise ValueError("TomTom result entry must be an object.")

    raw_address = payload.get("address")
    if not isinstance(raw_address, dict):
        raise ValueError("TomTom result address must be an object.")

    address = _coerce_optional_text(raw_address.get("freeformAddress"))
    if address is None:
        raise ValueError("TomTom result has no valid freeformAddress.")

    return AddressSuggestion(
        address=address,
        street=_coerce_optional_text(raw_address.get("streetName")),
        house=_coerce_optional_text(raw_address.get("streetNumber")),
        flat=None,
        postal_code=_coerce_optional_text(raw_address.get("postalCode")),
    )


def _coerce_optional_text(value: object) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise ValueError("TomTom address field must be a string.")
    return value.strip() or None