Merge pull request #233 from ricardogsilva/230-add-API-endpoint-for-s…

…erving-municipality-centroids-as-geojson

api endpoint for serving municipality centroids as geojson

arpav_ppcv/database.py

@@ -9,6 +9,7 @@
     Sequence,
 )
 
+import geojson_pydantic
 import shapely
 import shapely.io
 import sqlalchemy.exc
@@ -1256,6 +1257,63 @@ def generate_coverage_identifiers(
     return allowed_identifiers
 
 
+def list_municipality_centroids(
+    session: sqlmodel.Session,
+    *,
+    limit: int = 20,
+    offset: int = 0,
+    include_total: bool = False,
+    polygon_intersection_filter: shapely.Polygon = None,
+    name_filter: Optional[str] = None,
+    province_name_filter: Optional[str] = None,
+    region_name_filter: Optional[str] = None,
+) -> tuple[Sequence[municipalities.MunicipalityCentroid], Optional[int]]:
+    """List existing municipality centroids.
+
+    ``polygon_intersection_filter`` parameter is expected to be express a geometry in
+    the EPSG:4326 CRS.
+    """
+    statement = sqlmodel.select(municipalities.Municipality).order_by(
+        municipalities.Municipality.name
+    )
+    if name_filter is not None:
+        statement = _add_substring_filter(
+            statement, name_filter, municipalities.Municipality.name
+        )
+    if province_name_filter is not None:
+        statement = _add_substring_filter(
+            statement, province_name_filter, municipalities.Municipality.province_name
+        )
+    if region_name_filter is not None:
+        statement = _add_substring_filter(
+            statement, region_name_filter, municipalities.Municipality.region_name
+        )
+    if polygon_intersection_filter is not None:
+        statement = statement.where(
+            func.ST_Intersects(
+                municipalities.Municipality.geom,
+                func.ST_GeomFromWKB(
+                    shapely.io.to_wkb(polygon_intersection_filter), 4326
+                ),
+            )
+        )
+    items = session.exec(statement.offset(offset).limit(limit)).all()
+    num_items = _get_total_num_records(session, statement) if include_total else None
+    return [
+        municipalities.MunicipalityCentroid(
+            id=i.id,
+            name=i.name,
+            province_name=i.province_name,
+            region_name=i.region_name,
+            geom=geojson_pydantic.Point(
+                type="Point",
+                coordinates=(i.centroid_epsg_4326_lon, i.centroid_epsg_4326_lat),
+            ),
+        )
+        for i in items
+    ], num_items
+
+
 def list_municipalities(
     session: sqlmodel.Session,
     *,

arpav_ppcv/schemas/municipalities.py

@@ -36,3 +36,13 @@ class MunicipalityCreate(sqlmodel.SQLModel):
     region_name: str
     centroid_epsg_4326_lon: float | None = None
     centroid_epsg_4326_lat: float | None = None
+
+
+class MunicipalityCentroid(sqlmodel.SQLModel):
+    # model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+
+    id: pydantic.UUID4
+    geom: geojson_pydantic.Point
+    name: str
+    province_name: str
+    region_name: str

arpav_ppcv/webapp/api_v2/routers/municipalities.py

@@ -1,6 +1,7 @@
 import logging
 from typing import Annotated
 
+import pydantic
 import shapely.io
 from fastapi import (
     APIRouter,
@@ -20,7 +21,7 @@
 
 
 @router.get(
-    "/",
+    "/municipalities",
     response_class=GeoJsonResponse,
     response_model=municipalities_geojson.MunicipalityFeatureCollection,
 )
@@ -66,3 +67,49 @@ def list_municipalities(
         filtered_total=filtered_total,
         unfiltered_total=unfiltered_total,
     )
+
+
+@router.get(
+    "/municipality-centroids",
+    response_class=GeoJsonResponse,
+    response_model=municipalities_geojson.MunicipalityCentroidFeatureCollection,
+)
+def list_municipality_centroids(
+    request: Request,
+    db_session: Annotated[Session, Depends(dependencies.get_db_session)],
+    offset: Annotated[int, pydantic.Field(ge=0)] = 0,
+    limit: Annotated[int, pydantic.Field(ge=0, le=2000)] = 2000,
+    coords: str | None = None,
+    name: str | None = None,
+    province: str | None = None,
+    region: str | None = None,
+):
+    """List Italian municipality centroids."""
+    geom_filter_kwarg = {}
+    if coords is not None:
+        geom = shapely.io.from_wkt(coords)
+        if geom.geom_type == "Polygon":
+            geom_filter_kwarg = {"polygon_intersection_filter": geom}
+        else:
+            raise HTTPException(status_code=400, detail="geometry must be a polygon")
+    centroids, filtered_total = db.list_municipality_centroids(
+        db_session,
+        limit=limit,
+        offset=offset,
+        include_total=True,
+        name_filter=name,
+        province_name_filter=province,
+        region_name_filter=region,
+        **geom_filter_kwarg,
+    )
+    _, unfiltered_total = db.list_municipalities(
+        db_session, limit=1, offset=0, include_total=True
+    )
+    return municipalities_geojson.MunicipalityCentroidFeatureCollection.from_items(
+        centroids,
+        request,
+        limit=limit,
+        offset=offset,
+        filtered_total=filtered_total,
+        unfiltered_total=unfiltered_total,
+    )

arpav_ppcv/webapp/api_v2/schemas/geojson/municipalities.py

@@ -36,6 +36,38 @@ def from_db_instance(
         )
 
 
+class MunicipalityCentroidFeatureCollectionItem(geojson_pydantic.Feature):
+    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+
+    type: str = "Feature"
+    id: pydantic.UUID4
+    geometry: geojson_pydantic.Point
+
+    @classmethod
+    def from_db_instance(
+        cls,
+        instance: municipalities.Municipality,
+        request: Request,
+    ) -> "MunicipalityCentroidFeatureCollectionItem":
+        return cls(
+            id=instance.id,
+            geometry=instance.geom,
+            properties={
+                **instance.model_dump(
+                    exclude={
+                        "id",
+                        "geom",
+                    }
+                ),
+            },
+        )
+
+
 class MunicipalityFeatureCollection(ArpavFeatureCollection):
     path_operation_name = "list_municipalities"
     list_item_type = MunicipalityFeatureCollectionItem
+
+
+class MunicipalityCentroidFeatureCollection(ArpavFeatureCollection):
+    path_operation_name = "list_municipality_centroids"
+    list_item_type = MunicipalityCentroidFeatureCollectionItem