borderless / apis /rest_countries.py
spagestic's picture
langgraph integration
b23426c
Raw
History Blame Contribute Delete
3.66 kB
# apis/rest_countries.py
"""Country metadata from the REST Countries API with static fallback."""
from __future__ import annotations
from functools import lru_cache
from pathlib import Path
from typing import Any
import requests
import yaml
REST_COUNTRIES_URL = (
"https://restcountries.com/v3.1/all"
"?fields=name,cca2,cca3,region,subregion,capital,flags,latlng,area"
)
FALLBACK_PATH = Path(__file__).resolve().parents[1] / "data" / "countries_fallback.yaml"
@lru_cache(maxsize=1)
def _fallback_index() -> dict[str, dict[str, Any]]:
with FALLBACK_PATH.open(encoding="utf-8") as handle:
raw = yaml.safe_load(handle) or {}
by_iso2: dict[str, dict[str, Any]] = {}
by_iso3: dict[str, dict[str, Any]] = {}
for iso2, entry in raw.items():
normalized = {
"name": entry["name"],
"lat": float(entry["lat"]),
"lng": float(entry["lng"]),
"cca2": str(entry["cca2"]).upper(),
"cca3": str(entry["cca3"]).upper(),
"region": entry.get("region"),
"subregion": entry.get("subregion"),
"capital": entry.get("capital") or [],
"area": entry.get("area"),
"flag": entry.get("flag"),
"flag_alt": entry.get("flag_alt"),
}
by_iso2[normalized["cca2"]] = normalized
by_iso3[normalized["cca3"]] = normalized
return {"iso2": by_iso2, "iso3": by_iso3}
def _entry_from_api_country(country: dict[str, Any]) -> dict[str, Any] | None:
latlng = country.get("latlng")
if not latlng or len(latlng) != 2:
return None
lat, lng = latlng
if not isinstance(lat, (int, float)) or not isinstance(lng, (int, float)):
return None
return {
"name": country["name"]["common"],
"lat": float(lat),
"lng": float(lng),
"cca2": country.get("cca2", "").upper(),
"cca3": country.get("cca3", "").upper(),
"region": country.get("region"),
"subregion": country.get("subregion"),
"capital": country.get("capital") or [],
"area": country.get("area"),
"flag": (country.get("flags") or {}).get("png")
or (country.get("flags") or {}).get("svg"),
"flag_alt": (country.get("flags") or {}).get("alt"),
}
@lru_cache(maxsize=1)
def _country_index() -> dict[str, dict[str, Any]]:
by_iso2: dict[str, dict[str, Any]] = {}
by_iso3: dict[str, dict[str, Any]] = {}
try:
resp = requests.get(REST_COUNTRIES_URL, timeout=20)
resp.raise_for_status()
payload = resp.json()
if isinstance(payload, list):
for country in payload:
if not isinstance(country, dict):
continue
entry = _entry_from_api_country(country)
if not entry:
continue
if entry["cca2"]:
by_iso2[entry["cca2"]] = entry
if entry["cca3"]:
by_iso3[entry["cca3"]] = entry
except Exception:
pass
if by_iso2:
return {"iso2": by_iso2, "iso3": by_iso3}
return _fallback_index()
def lookup_country(code: str) -> dict[str, Any] | None:
normalized = (code or "").strip().upper()
if not normalized:
return None
index = _country_index()
if len(normalized) == 2:
return index["iso2"].get(normalized)
if len(normalized) == 3:
return index["iso3"].get(normalized)
return index["iso2"].get(normalized) or index["iso3"].get(normalized)