Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 62 additions & 0 deletions geocoding.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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] = {}
Expand Down Expand Up @@ -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:
Expand Down
31 changes: 29 additions & 2 deletions service.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.")
Expand Down
52 changes: 50 additions & 2 deletions tests/service_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Loading