👕 style: black formatting

Author: gtempus

app/crud/geostore.py

@@ -192,17 +192,27 @@ async def get_first_row(sql: Select):
 
 
 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,
+    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"),]
-    row = await _find_first_geostore(adm_level, admin_provider, admin_version, columns_etc, country_id, region_id,
-                               src_table, subregion_id)
+    columns_etc: List[Column | Label] = [
+        db.column("gfw_geostore_id"),
+    ]
+    row = await _find_first_geostore(
+        adm_level,
+        admin_provider,
+        admin_version,
+        columns_etc,
+        country_id,
+        region_id,
+        src_table,
+        subregion_id,
+    )
     return await row.gfw_geostore_id
 
 
@@ -240,8 +250,16 @@ async def build_gadm_geostore(
             )
         )
 
-    row = await _find_first_geostore(adm_level, admin_provider, admin_version, columns_etc, country_id, region_id,
-                                     src_table, subregion_id)
+    row = await _find_first_geostore(
+        adm_level,
+        admin_provider,
+        admin_version,
+        columns_etc,
+        country_id,
+        region_id,
+        src_table,
+        subregion_id,
+    )
 
     if row.geojson is None:
         raise GeometryIsNullError(
@@ -261,10 +279,26 @@ async def build_gadm_geostore(
     )
 
 
-async def _find_first_geostore(adm_level, admin_provider, admin_version, columns_etc, country_id, region_id, src_table,
-                               subregion_id):
+async def _find_first_geostore(
+    adm_level,
+    admin_provider,
+    admin_version,
+    columns_etc,
+    country_id,
+    region_id,
+    src_table,
+    subregion_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)
+    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(
@@ -273,7 +307,9 @@ async def _find_first_geostore(adm_level, admin_provider, admin_version, columns
     return row
 
 
-async def add_where_clauses(adm_level, admin_provider, admin_version, country_id, region_id, sql, subregion_id):
+async def add_where_clauses(
+    adm_level, admin_provider, admin_version, country_id, region_id, sql, subregion_id
+):
     where_clauses: List[TextClause] = [
         db.text("adm_level=:adm_level").bindparams(adm_level=str(adm_level))
     ]
@@ -335,7 +371,7 @@ async def get_gadm_geostore(
         admin_version=admin_version,
         adm_level=adm_level,
         simplify=simplify,
-    country_id=country_id,
+        country_id=country_id,
         region_id=region_id,
         subregion_id=subregion_id,
     )

app/models/pydantic/datamart.py

@@ -1,14 +1,14 @@
+from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Dict, Literal, Optional, Union
 from uuid import UUID
-from abc import ABC, abstractmethod
 
 from pydantic import Field, root_validator, validator
 
 from app.models.pydantic.responses import Response
 
-from .base import StrictBaseModel
 from ...crud.geostore import get_gadm_geostore_id
+from .base import StrictBaseModel
 
 
 class AreaOfInterest(StrictBaseModel, ABC):
@@ -19,24 +19,30 @@ async def get_geostore_id(self) -> UUID:
 
 
 class GeostoreAreaOfInterest(AreaOfInterest):
-    type: Literal['geostore'] = 'geostore'
-    geostore_id:UUID = Field(..., title="Geostore ID")
+    type: Literal["geostore"] = "geostore"
+    geostore_id: UUID = Field(..., title="Geostore ID")
 
     async def get_geostore_id(self) -> UUID:
         return self.geostore_id
 
 
 class AdminAreaOfInterest(AreaOfInterest):
-    type: Literal['admin'] = 'admin'
+    type: Literal["admin"] = "admin"
     country: str = Field(..., title="ISO Country Code")
     region: Optional[str] = Field(None, title="Region")
     subregion: Optional[str] = Field(None, title="Subregion")
-    provider: str = Field('gadm', title="Administrative Boundary Provider")
-    version: str = Field('4.1', title="Administrative Boundary Version")
-
+    provider: str = Field("gadm", title="Administrative Boundary Provider")
+    version: str = Field("4.1", title="Administrative Boundary Version")
 
     async def get_geostore_id(self) -> UUID:
-        admin_level = sum(1 for field in (self.country, self.region, self.subregion) if field is not None) - 1
+        admin_level = (
+            sum(
+                1
+                for field in (self.country, self.region, self.subregion)
+                if field is not None
+            )
+            - 1
+        )
         geostore_id = await get_gadm_geostore_id(
             admin_provider=self.provider,
             admin_version=self.version,
@@ -55,14 +61,13 @@ def check_region_subregion(cls, values):
             raise ValueError("region must be specified if subregion is provided")
         return values
 
-    @validator('provider', pre=True, always=True)
+    @validator("provider", pre=True, always=True)
     def set_provider_default(cls, v):
-        return v or 'gadm'
+        return v or "gadm"
 
-    @validator('version', pre=True, always=True)
+    @validator("version", pre=True, always=True)
     def set_version_default(cls, v):
-        return v or '4.1'
-
+        return v or "4.1"
 
 
 class AnalysisStatus(str, Enum):
@@ -99,7 +104,9 @@ class DataMartResourceLinkResponse(Response):
 
 
 class TreeCoverLossByDriverIn(StrictBaseModel):
-    aoi: Union[GeostoreAreaOfInterest, AdminAreaOfInterest] = Field(..., discriminator='type')
+    aoi: Union[GeostoreAreaOfInterest, AdminAreaOfInterest] = Field(
+        ..., discriminator="type"
+    )
     canopy_cover: int = 30
     dataset_version: Dict[str, str] = {}
 
@@ -132,6 +139,3 @@ class Config:
 
 class TreeCoverLossByDriverResponse(Response):
     data: TreeCoverLossByDriver
-
-
-

app/routes/datamart/__init__.py

@@ -13,7 +13,7 @@
                     "value": {
                         "type": "geostore",
                         "geostore_id": "637d378f-93a9-4364-bfa8-95b6afd28c3a",
-                    }
+                    },
                 },
                 "Admin Area Of Interest": {
                     "summary": "Admin Area Of Interest",
@@ -23,16 +23,16 @@
                         "country": "BRA",
                         "region": "12",
                         "subregion": "2",
-                    }
-                }
+                    },
+                },
             },
             "description": "The Area of Interest",
             "schema": {
                 "oneOf": [
                     {"$ref": "#/components/schemas/GeostoreAreaOfInterest"},
                     {"$ref": "#/components/schemas/AdminAreaOfInterest"},
                 ]
-            }
+            },
         },
         {
             "name": "dataset_version",
@@ -50,7 +50,7 @@
             },
             "description": (
                 "Pass dataset version overrides as bracketed query parameters.",
-            )
-        }
+            ),
+        },
     ]
-}
+}

app/routes/datamart/land.py

@@ -23,29 +23,30 @@
 from app.errors import RecordNotFoundError
 from app.models.enum.geostore import GeostoreOrigin
 from app.models.pydantic.datamart import (
+    AdminAreaOfInterest,
     AnalysisStatus,
+    AreaOfInterest,
     DataMartResource,
     DataMartResourceLink,
     DataMartResourceLinkResponse,
+    GeostoreAreaOfInterest,
     TreeCoverLossByDriver,
     TreeCoverLossByDriverIn,
     TreeCoverLossByDriverResponse,
-    AreaOfInterest,
-    GeostoreAreaOfInterest,
-    AdminAreaOfInterest,
 )
 from app.settings.globals import API_URL
 from app.tasks.datamart.land import (
     DEFAULT_LAND_DATASET_VERSIONS,
     compute_tree_cover_loss_by_driver,
 )
 from app.utils.geostore import get_geostore
-from . import OPENAPI_EXTRA
 
 from ...authentication.api_keys import get_api_key
+from . import OPENAPI_EXTRA
 
 router = APIRouter()
 
+
 def _parse_dataset_versions(request: Request) -> Dict[str, str]:
     dataset_versions = {}
     errors = []
@@ -69,23 +70,27 @@ def _parse_dataset_versions(request: Request) -> Dict[str, str]:
 
 def _parse_area_of_interest(request: Request) -> AreaOfInterest:
     params = request.query_params
-    aoi_type = params.get('aoi[type]')
+    aoi_type = params.get("aoi[type]")
     try:
-        if aoi_type == 'geostore':
-            return GeostoreAreaOfInterest(geostore_id=params.get('aoi[geostore_id]', None))
+        if aoi_type == "geostore":
+            return GeostoreAreaOfInterest(
+                geostore_id=params.get("aoi[geostore_id]", None)
+            )
 
             # Otherwise, check if the request contains admin area information
-        if aoi_type == 'admin':
+        if aoi_type == "admin":
             return AdminAreaOfInterest(
-                country=params.get('aoi[country]', None),
-                region=params.get('aoi[region]', None),
-                subregion=params.get('aoi[subregion]', None),
-                provider=params.get('aoi[provider]', None),
-                version=params.get('aoi[version]', None),
+                country=params.get("aoi[country]", None),
+                region=params.get("aoi[region]", None),
+                subregion=params.get("aoi[subregion]", None),
+                provider=params.get("aoi[provider]", None),
+                version=params.get("aoi[version]", None),
             )
 
         # If neither type is provided, raise an error
-        raise HTTPException(status_code=422, detail="Invalid Area of Interest parameters")
+        raise HTTPException(
+            status_code=422, detail="Invalid Area of Interest parameters"
+        )
     except ValidationError as e:
         raise HTTPException(status_code=422, detail=e.errors())
 
@@ -96,7 +101,7 @@ def _parse_area_of_interest(request: Request) -> AreaOfInterest:
     response_model=DataMartResourceLinkResponse,
     tags=["Land"],
     status_code=200,
-    openapi_extra= OPENAPI_EXTRA,
+    openapi_extra=OPENAPI_EXTRA,
 )
 async def tree_cover_loss_by_driver_search(
     *,

tests_v2/unit/app/crud/test_geostore.py

@@ -9,7 +9,7 @@
 
 @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"
@@ -82,7 +82,7 @@ async def test_get_gadm_geostore_generates_correct_sql_for_region_lookup(
 
 @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"
@@ -121,7 +121,7 @@ async def test_get_gadm_geostore_generates_correct_sql_for_subregion_lookup(
 class TestGadmGeostoreIDLookup:
     @pytest.mark.asyncio
     async def test_get_gadm_geostore_id_generates_correct_sql_for_country_lookup(
-            async_client: AsyncClient
+        async_client: AsyncClient,
     ):
         provider = "gadm"
         version = "4.1"
@@ -152,10 +152,9 @@ async def test_get_gadm_geostore_id_generates_correct_sql_for_country_lookup(
         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
+        async_client: AsyncClient,
     ):
         provider = "gadm"
         version = "4.1"
@@ -187,10 +186,9 @@ async def test_get_gadm_geostore_id_generates_correct_sql_for_region_lookup(
         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
+        async_client: AsyncClient,
     ):
         provider = "gadm"
         version = "4.1"