Раис Юсупалиев
160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
"""Dadata address suggestion adapter."""
|
|
|
|
from typing import Any
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
from app.adapters.address_suggestions.base import (
|
|
AddressSuggestionClientError,
|
|
AddressSuggestionProvider,
|
|
AddressSuggestionRequestError,
|
|
)
|
|
from app.config import DadataAddressSuggestionsConfig
|
|
from app.schemas.request import AddressSuggestRequest
|
|
from app.schemas.response import AddressSuggestion
|
|
|
|
|
|
class DadataClientError(AddressSuggestionClientError):
|
|
"""Raised when Dadata address suggestions fail."""
|
|
|
|
|
|
class DadataRequestError(DadataClientError, AddressSuggestionRequestError):
|
|
"""Raised when Dadata rejects a suggestion request."""
|
|
|
|
|
|
class DadataAddressSuggestionProvider(AddressSuggestionProvider):
|
|
name = "dadata"
|
|
|
|
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
|
|
self._api_key = api_key
|
|
self._timeout_seconds = timeout_seconds
|
|
|
|
@classmethod
|
|
def from_config(
|
|
cls,
|
|
*,
|
|
http_client: httpx.AsyncClient,
|
|
config: DadataAddressSuggestionsConfig,
|
|
) -> "DadataAddressSuggestionProvider":
|
|
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]:
|
|
payload = self._build_payload(request)
|
|
try:
|
|
response = await self._http_client.post(
|
|
self._url,
|
|
json=payload,
|
|
headers={
|
|
"Authorization": f"Token {self._api_key}",
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
},
|
|
timeout=self._timeout_seconds,
|
|
)
|
|
except (httpx.TimeoutException, httpx.TransportError) as exc:
|
|
raise DadataClientError(
|
|
"Dadata address suggestion request failed."
|
|
) from exc
|
|
|
|
if 400 <= response.status_code < 500:
|
|
raise DadataRequestError(
|
|
"Dadata address suggestion request was rejected with status "
|
|
f"{response.status_code}."
|
|
)
|
|
|
|
if response.status_code >= 500:
|
|
raise DadataClientError(
|
|
"Dadata address suggestion 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 DadataClientError(
|
|
"Dadata address suggestion returned invalid payload."
|
|
) from exc
|
|
|
|
try:
|
|
return _map_dadata_response(raw_payload)
|
|
except ValueError as exc:
|
|
raise DadataClientError(
|
|
"Dadata address suggestion response payload is invalid."
|
|
) from exc
|
|
|
|
@staticmethod
|
|
def _build_payload(request: AddressSuggestRequest) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"query": f"{request.city.strip()} {request.query.strip()}".strip(),
|
|
}
|
|
if request.limit is not None:
|
|
payload["count"] = request.limit
|
|
if request.country_code.strip().upper() != "RU":
|
|
payload["locations"] = [{"country": "*"}]
|
|
return payload
|
|
|
|
|
|
def _map_dadata_response(payload: object) -> list[AddressSuggestion]:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Dadata response must be a JSON object.")
|
|
|
|
raw_suggestions = payload.get("suggestions")
|
|
if not isinstance(raw_suggestions, list):
|
|
raise ValueError("Dadata response must include suggestions list.")
|
|
|
|
return [_map_dadata_suggestion(item) for item in raw_suggestions]
|
|
|
|
|
|
def _map_dadata_suggestion(payload: object) -> AddressSuggestion:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Dadata suggestion entry must be an object.")
|
|
|
|
raw_address = payload.get("unrestricted_value") or payload.get("value")
|
|
if not isinstance(raw_address, str) or not raw_address.strip():
|
|
raise ValueError("Dadata suggestion has no valid address value.")
|
|
|
|
raw_data = payload.get("data")
|
|
if raw_data is not None and not isinstance(raw_data, dict):
|
|
raise ValueError("Dadata suggestion data must be an object when provided.")
|
|
|
|
street = None
|
|
house = None
|
|
flat = None
|
|
postal_code = None
|
|
if isinstance(raw_data, dict):
|
|
street = _coerce_optional_text(raw_data.get("street"))
|
|
house = _coerce_optional_text(raw_data.get("house"))
|
|
flat = _coerce_optional_text(raw_data.get("flat"))
|
|
postal_code = _coerce_optional_text(raw_data.get("postal_code"))
|
|
|
|
return AddressSuggestion(
|
|
address=raw_address.strip(),
|
|
street=street,
|
|
house=house,
|
|
flat=flat,
|
|
postal_code=postal_code,
|
|
)
|
|
|
|
|
|
def _coerce_optional_text(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
|
|
return str(value).strip() or None
|