feat(Geostore): add ability to retrieve geostore ID only

Author: gtempus

app/crud/geostore.py

@@ -191,6 +191,27 @@ async def get_first_row(sql: Select):
     return row
 
 
+async def get_gadm_geostore_id(
+        admin_provider: str,
+        admin_version: str,
+        adm_level: int,
+        country_id: str,
+        region_id: str | None = None,
+        subregion_id: str | None = None,
+) -> str:
+    src_table = await get_versioned_dataset(admin_provider, admin_version)
+    columns_etc: List[Column | Label] = [db.column("gfw_geostore_id"),]
+    sql: Select = db.select(columns_etc).select_from(src_table)
+    sql = await add_where_clauses(adm_level, admin_provider, admin_version, country_id, region_id, sql, subregion_id)
+    row = await get_first_row(sql)
+    if row is None:
+        raise RecordNotFoundError(
+            f"Admin boundary not found in {admin_provider} version {admin_version}"
+        )
+
+    return await row.gfw_geostore_id
+
+
 async def build_gadm_geostore(
     admin_provider: str,
     admin_version: str,

tests_v2/unit/app/crud/test_geostore.py

@@ -3,13 +3,13 @@
 import pytest
 from httpx import AsyncClient
 
-from app.crud.geostore import get_gadm_geostore
+from app.crud.geostore import get_gadm_geostore, get_gadm_geostore_id
 from app.errors import RecordNotFoundError
 
 
 @pytest.mark.asyncio
 async def test_get_gadm_geostore_generates_correct_sql_for_country_lookup(
-    async_client: AsyncClient
+        async_client: AsyncClient
 ):
     provider = "gadm"
     version = "4.1"
@@ -45,7 +45,7 @@ async def test_get_gadm_geostore_generates_correct_sql_for_country_lookup(
 
 @pytest.mark.asyncio
 async def test_get_gadm_geostore_generates_correct_sql_for_region_lookup(
-    async_client: AsyncClient
+        async_client: AsyncClient
 ):
     provider = "gadm"
     version = "4.1"
@@ -80,10 +80,9 @@ async def test_get_gadm_geostore_generates_correct_sql_for_region_lookup(
     assert actual_sql == expected_sql
 
 
-
 @pytest.mark.asyncio
 async def test_get_gadm_geostore_generates_correct_sql_for_subregion_lookup(
-    async_client: AsyncClient
+        async_client: AsyncClient
 ):
     provider = "gadm"
     version = "4.1"
@@ -117,3 +116,109 @@ async def test_get_gadm_geostore_generates_correct_sql_for_subregion_lookup(
 
     assert mock_get_first_row.called is True
     assert actual_sql == expected_sql
+
+
+class TestGadmGeostoreIDLookup:
+    @pytest.mark.asyncio
+    async def test_get_gadm_geostore_id_generates_correct_sql_for_country_lookup(
+            async_client: AsyncClient
+    ):
+        provider = "gadm"
+        version = "4.1"
+        adm_level = 0
+        country_id = "MEX"
+
+        with patch("app.crud.geostore.get_first_row") as mock_get_first_row:
+            mock_get_first_row.return_value = None
+            try:
+                _ = await get_gadm_geostore_id(
+                    provider, version, adm_level, country_id, None, None
+                )
+            except RecordNotFoundError:
+                pass
+
+        expected_sql = (
+            "SELECT gfw_geostore_id "
+            '\nFROM gadm_administrative_boundaries."v4.1.64" \n'
+            "WHERE adm_level='0' AND gid_0='MEX'"
+        )
+
+        actual_sql = str(
+            mock_get_first_row.call_args.args[0].compile(
+                compile_kwargs={"literal_binds": True}
+            )
+        )
+
+        assert mock_get_first_row.called is True
+        assert actual_sql == expected_sql
+
+
+    @pytest.mark.asyncio
+    async def test_get_gadm_geostore_id_generates_correct_sql_for_region_lookup(
+            async_client: AsyncClient
+    ):
+        provider = "gadm"
+        version = "4.1"
+        adm_level = 1
+        country = "MEX"
+        region = "5"
+
+        with patch("app.crud.geostore.get_first_row") as mock_get_first_row:
+            mock_get_first_row.return_value = None
+            try:
+                _ = await get_gadm_geostore_id(
+                    provider, version, adm_level, country, region, None
+                )
+            except RecordNotFoundError:
+                pass
+
+        expected_sql = (
+            "SELECT gfw_geostore_id "
+            '\nFROM gadm_administrative_boundaries."v4.1.64" \n'
+            r"WHERE adm_level='1' AND gid_1 LIKE 'MEX.5\__'"
+        )
+
+        actual_sql = str(
+            mock_get_first_row.call_args.args[0].compile(
+                compile_kwargs={"literal_binds": True}
+            )
+        )
+
+        assert mock_get_first_row.called is True
+        assert actual_sql == expected_sql
+
+
+    @pytest.mark.asyncio
+    async def test_get_gadm_geostore_id_generates_correct_sql_for_subregion_lookup(
+            async_client: AsyncClient
+    ):
+        provider = "gadm"
+        version = "4.1"
+        adm_level = 2
+        country = "MEX"
+        region = "5"
+        subregion = "2"
+
+        with patch("app.crud.geostore.get_first_row") as mock_get_first_row:
+            mock_get_first_row.return_value = None
+            try:
+                _ = await get_gadm_geostore_id(
+                    provider, version, adm_level, country, region, subregion
+                )
+            except RecordNotFoundError:
+                pass
+
+        expected_sql = (
+            "SELECT gfw_geostore_id "
+            '\nFROM gadm_administrative_boundaries."v4.1.64" \n'
+            r"WHERE adm_level='2' AND gid_2 LIKE 'MEX.5.2\__'"
+        )
+
+        actual_sql = str(
+            mock_get_first_row.call_args.args[0].compile(
+                compile_kwargs={"literal_binds": True}
+            )
+        )
+
+        assert mock_get_first_row.called is True
+        assert actual_sql == expected_sql