Too much, TBH: Add limiting query to specified admin level; enforce s…

…pecifying admin level hierarchy

app/routes/thematic/geoencoder.py

@@ -1,5 +1,5 @@
 import re
-from typing import Optional, Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, HTTPException, Query
 
@@ -20,7 +20,7 @@
 )
 async def geoencode(
     *,
-    admin_source: Optional[str] = Query(
+    admin_source: str = Query(
         "GADM",
         description="The source of administrative boundaries to use."
     ),
@@ -43,7 +43,7 @@ async def geoencode(
     """ Look up administrative boundary IDs matching a specified country name
     (and region name and subregion names, if specified).
     """
-    admin_source_to_dataset = {
+    admin_source_to_dataset: Dict[str, str] = {
         "GADM": "gadm_administrative_boundaries"
     }
 
@@ -52,13 +52,18 @@ async def geoencode(
     except KeyError:
         raise HTTPException(
             status_code=400,
-            detail=f"Invalid admin boundary source. Valid sources: {admin_source_to_dataset.keys()}"
+            detail=(
+                "Invalid admin boundary source. Valid sources:"
+                f" {[source for source in admin_source_to_dataset.keys()]}"
+            )
         )
 
     version_str = "v" + str(admin_version).lstrip("v")
 
     await version_is_valid(dataset, version_str)
 
+    ensure_admin_hierarchy(region, subregion)
+
     sql: str = _admin_boundary_lookup_sql(
         admin_source,
         country,
@@ -79,11 +84,33 @@ async def geoencode(
     )
 
 
+def ensure_admin_hierarchy(region: str | None, subregion: str | None) -> None:
+    if subregion and not region:
+        raise HTTPException(
+            status_code=400,
+            detail= "If subregion is specified, region must be specified as well."
+        )
+
+
+def determine_admin_level(country: str | None, region: str | None, subregion: str | None) -> str:
+    if subregion:
+        return "2"
+    elif region:
+        return "1"
+    elif country:
+        return "0"
+    else:  # Shouldn't get here if FastAPI route definition worked
+        raise HTTPException(
+            status_code=400,
+            detail= "Country MUST be specified."
+        )
+
+
 def _admin_boundary_lookup_sql(
     dataset: str,
     country_name: str,
-    region_name: Optional[str],
-    subregion_name: Optional[str]
+    region_name: str | None,
+    subregion_name: str | None
 ):
     """Generate the SQL required to look up administrative boundary
     IDs by name.
@@ -97,6 +124,9 @@ def _admin_boundary_lookup_sql(
     if subregion_name is not None:
         sql += f" AND name_2='{subregion_name}'"
 
+    adm_level = determine_admin_level(country_name, region_name, subregion_name)
+    sql += f" AND adm_level='{adm_level}'"
+
     return sql
 
 

tests_v2/unit/app/routes/thematic/geoencoder/test_geoencoder.py

@@ -15,7 +15,7 @@ async def test__admin_boundary_lookup_sql_country() -> None:
     )
     assert sql == (
         "SELECT gid_0, gid_1, gid_2, country, name_1, name_2 FROM some_dataset"
-        " WHERE country='some_country'"
+        " WHERE country='some_country' AND adm_level='0'"
     )
 
 
@@ -28,6 +28,7 @@ async def test__admin_boundary_lookup_sql_country_region() -> None:
         "SELECT gid_0, gid_1, gid_2, country, name_1, name_2 FROM some_dataset"
         " WHERE country='some_country'"
         " AND name_1='some_region'"
+        " AND adm_level='1'"
     )
 
 
@@ -41,6 +42,7 @@ async def test__admin_boundary_lookup_sql_all() -> None:
         " WHERE country='some_country'"
         " AND name_1='some_region'"
         " AND name_2='some_subregion'"
+        " AND adm_level='2'"
     )