From e7666cd1e543f2ab47fffd02145e82858a8d3150 Mon Sep 17 00:00:00 2001 From: rup-narayan-rajbanshi Date: Thu, 3 Sep 2026 16:03:36 +0545 Subject: [PATCH] feat(iso3): add func to get iso3 from iso2 and country_name & vice-versa --- geocoding.py | 62 +++++++++++++++++++++++++++++++++++++++++++ service.py | 31 ++++++++++++++++++++-- tests/service_test.py | 52 ++++++++++++++++++++++++++++++++++-- 3 files changed, 141 insertions(+), 4 deletions(-) diff --git a/geocoding.py b/geocoding.py index 6fb5390..0bae8a1 100644 --- a/geocoding.py +++ b/geocoding.py @@ -20,6 +20,10 @@ class AdminGeometry(pydantic.BaseModel): geometry: dict[str, typing.Any] +class Iso3Response(pydantic.BaseModel): + iso3: str + + class FastGeocoder: _wab_path: str _gaul_path: str @@ -32,6 +36,9 @@ def __init__(self, wab_path: str, gaul_path: str) -> None: self._geom_from_country_name_cache: dict[str, AdminGeometry] = {} self._geom_from_iso3_cache: dict[str, AdminGeometry] = {} self._geom_from_adm_names_cache: dict[str, AdminGeometry] = {} + self._iso3_from_country_name_cache: dict[str, str] = {} + self._iso3_from_iso2_cache: dict[str, str] = {} + self._country_from_iso3_cache: dict[str, Country] = {} # gaul self._adm1_to_geometry_mapping: dict[int, BaseGeometry] = {} @@ -68,6 +75,61 @@ def get_iso3_from_geometry(self, lng: float, lat: float) -> Country | None: ) return None + def get_iso3_from_country_name(self, country_name: str) -> str | None: + key = country_name.lower().strip() + from_cache = self._iso3_from_country_name_cache.get(key) + if from_cache: + return from_cache + + with fiona.open(self._wab_path, layer=WAB_LAYER) as src: + for feature in src: + properties: dict[str, typing.Any] = feature["properties"] + if properties["name"].lower().strip() == key: + iso3 = properties["iso3"] + self._iso3_from_country_name_cache[key] = iso3 + return iso3 + return None + + def get_iso3_from_iso2(self, iso2: str) -> str | None: + key = iso2.lower().strip() + from_cache = self._iso3_from_iso2_cache.get(key) + if from_cache: + return from_cache + + with fiona.open(self._wab_path, layer=WAB_LAYER) as src: + for feature in src: + properties: dict[str, typing.Any] = feature["properties"] + iso2_from_feature = properties["iso_3166_1_alpha_2_codes"] + if not iso2_from_feature: + continue + if iso2_from_feature.lower().strip() == key: + iso3 = properties["iso3"] + self._iso3_from_iso2_cache[key] = iso3 + return iso3 + return None + + def get_country_from_iso3(self, iso3: str) -> Country | None: + key = iso3.lower().strip() + from_cache = self._country_from_iso3_cache.get(key) + if from_cache: + return from_cache + + with fiona.open(self._wab_path, layer=WAB_LAYER) as src: + for feature in src: + properties: dict[str, typing.Any] = feature["properties"] + iso3_from_feature = properties["iso3"] + if not iso3_from_feature: + continue + if iso3_from_feature.lower().strip() == key: + country = Country( + name=properties["name"], + iso3=iso3_from_feature, + iso2=properties["iso_3166_1_alpha_2_codes"], + ) + self._country_from_iso3_cache[key] = country + return country + return None + def get_geometry_from_country_name(self, country_name: str) -> AdminGeometry | None: from_cache = self._geom_from_country_name_cache.get(country_name) if from_cache: diff --git a/service.py b/service.py index fdc13dd..077e164 100644 --- a/service.py +++ b/service.py @@ -42,12 +42,39 @@ async def home(): @app.get("/country/iso3") -async def get_iso3(lat: float, lng: float) -> geocoding.Country: - """Get the iso3 based on coordinate""" +async def get_iso3( + lat: float | None = None, + lng: float | None = None, + country_name: str | None = None, + iso2: str | None = None, + iso3: str | None = None, +) -> geocoding.Country | geocoding.Iso3Response: + """Get country info based on coordinate, country name, ISO2, or ISO3 code""" try: geocoder = shared_mem["geocoder"] if not geocoder: raise Exception("Geocoder is not initialized") + + if country_name: + result = geocoder.get_iso3_from_country_name(country_name) + if not result: + raise HTTPException(status_code=404, detail="iso3 not found.") + return geocoding.Iso3Response(iso3=result) + + if iso2: + result = geocoder.get_iso3_from_iso2(iso2) + if not result: + raise HTTPException(status_code=404, detail="iso3 not found.") + return geocoding.Iso3Response(iso3=result) + + if iso3: + result = geocoder.get_country_from_iso3(iso3) + if not result: + raise HTTPException(status_code=404, detail="Country not found.") + return result + + if lat is None or lng is None: + raise HTTPException(status_code=400, detail="Provide lat/lng, country_name, iso2, or iso3.") result = geocoder.get_iso3_from_geometry(lng=lng, lat=lat) if not result: raise HTTPException(status_code=404, detail="iso3 not found.") diff --git a/tests/service_test.py b/tests/service_test.py index e051599..e8fc216 100644 --- a/tests/service_test.py +++ b/tests/service_test.py @@ -69,10 +69,58 @@ def test_returns_500_on_unexpected_error(self, mock_geocoder): assert response.status_code == 500 - def test_requires_lat_and_lng_query_params(self): + def test_returns_400_when_no_params_given(self, mock_geocoder): response = client.get("/country/iso3") - assert response.status_code == 422 + assert response.status_code == 400 + + def test_returns_iso3_by_country_name(self, mock_geocoder): + mock_geocoder.get_iso3_from_country_name.return_value = "NPL" + + response = client.get("/country/iso3", params={"country_name": "Nepal"}) + + assert response.status_code == 200 + assert response.json() == {"iso3": "NPL"} + mock_geocoder.get_iso3_from_country_name.assert_called_once_with("Nepal") + + def test_returns_404_when_country_name_not_found(self, mock_geocoder): + mock_geocoder.get_iso3_from_country_name.return_value = None + + response = client.get("/country/iso3", params={"country_name": "Atlantis"}) + + assert response.status_code == 404 + + def test_returns_iso3_by_iso2(self, mock_geocoder): + mock_geocoder.get_iso3_from_iso2.return_value = "FRA" + + response = client.get("/country/iso3", params={"iso2": "FR"}) + + assert response.status_code == 200 + assert response.json() == {"iso3": "FRA"} + mock_geocoder.get_iso3_from_iso2.assert_called_once_with("FR") + + def test_returns_404_when_iso2_not_found(self, mock_geocoder): + mock_geocoder.get_iso3_from_iso2.return_value = None + + response = client.get("/country/iso3", params={"iso2": "XX"}) + + assert response.status_code == 404 + + def test_returns_country_by_iso3(self, mock_geocoder): + mock_geocoder.get_country_from_iso3.return_value = Country(name="Nepal", iso3="NPL", iso2="NP") + + response = client.get("/country/iso3", params={"iso3": "NPL"}) + + assert response.status_code == 200 + assert response.json() == {"name": "Nepal", "iso3": "NPL", "iso2": "NP"} + mock_geocoder.get_country_from_iso3.assert_called_once_with("NPL") + + def test_returns_404_when_iso3_not_found(self, mock_geocoder): + mock_geocoder.get_country_from_iso3.return_value = None + + response = client.get("/country/iso3", params={"iso3": "ZZZ"}) + + assert response.status_code == 404 class TestGetCountryGeometry: