Добавлен tomtom

2026-04-03 04:19:55 +03:00
parent 44893e01ae
commit db1c74f1ce
16 changed files with 741 additions and 30 deletions
@@ -0,0 +1,155 @@
+"""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