Merge pull request #137 from ricardogsilva/132-add-API-endpoint-to-fi…

…nd-municipality-from-coordinates

Find municipality by coordinates

arpav_ppcv/database.py

@@ -9,7 +9,7 @@
     Sequence,
 )
 
-import shapely.geometry
+import shapely
 import shapely.io
 import sqlalchemy.exc
 import sqlmodel
@@ -1024,6 +1024,57 @@ def list_allowed_coverage_identifiers(
     return result
 
 
+def list_municipalities(
+    session: sqlmodel.Session,
+    *,
+    limit: int = 20,
+    offset: int = 0,
+    include_total: bool = False,
+    polygon_intersection_filter: shapely.Polygon = None,
+    point_filter: shapely.Point = None,
+    name_filter: Optional[str] = None,
+    province_name_filter: Optional[str] = None,
+    region_name_filter: Optional[str] = None,
+) -> Optional[municipalities.Municipality]:
+    """List existing municipalities.
+
+    Both ``polygon_intersection_filter`` and ``point_filter`` parameters are expected
+    to be a geometries in the EPSG:4326 CRS.
+    """
+    statement = sqlmodel.select(municipalities.Municipality).order_by(
+        municipalities.Municipality.name
+    )
+    if name_filter is not None:
+        statement = statement.where(municipalities.Municipality.name.ilike(name_filter))
+    if province_name_filter is not None:
+        statement = statement.where(
+            municipalities.Municipality.province_name.ilike(province_name_filter)
+        )
+    if region_name_filter is not None:
+        statement = statement.where(
+            municipalities.Municipality.region_name.ilike(region_name_filter)
+        )
+    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
+                ),
+            )
+        )
+    if point_filter is not None:
+        statement = statement.where(
+            func.ST_Intersects(
+                municipalities.Municipality.geom,
+                func.ST_GeomFromWKB(shapely.io.to_wkb(point_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 items, num_items
+
+
 def create_many_municipalities(
     session: sqlmodel.Session,
     municipalities_to_create: Sequence[municipalities.MunicipalityCreate],

arpav_ppcv/webapp/api_v2/app.py

@@ -3,6 +3,7 @@
 
 from ... import config
 from .routers.coverages import router as coverages_router
+from .routers.municipalities import router as municipalities_router
 from .routers.observations import router as observations_router
 from .routers.base import router as base_router
 
@@ -50,4 +51,11 @@ def create_app(settings: config.ArpavPpcvSettings) -> fastapi.FastAPI:
             "observations",
         ],
     )
+    app.include_router(
+        municipalities_router,
+        prefix="/municipalities",
+        tags=[
+            "municipalities",
+        ],
+    )
     return app

arpav_ppcv/webapp/api_v2/routers/municipalities.py

@@ -0,0 +1,68 @@
+import logging
+from typing import Annotated
+
+import shapely.io
+from fastapi import (
+    APIRouter,
+    Depends,
+    HTTPException,
+    Request,
+)
+from sqlmodel import Session
+
+from .... import database as db
+from ... import dependencies
+from ...responses import GeoJsonResponse
+from ..schemas.geojson import municipalities as municipalities_geojson
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+
+
+@router.get(
+    "/",
+    response_class=GeoJsonResponse,
+    response_model=municipalities_geojson.MunicipalityFeatureCollection,
+)
+def list_municipalities(
+    request: Request,
+    db_session: Annotated[Session, Depends(dependencies.get_db_session)],
+    list_params: Annotated[dependencies.CommonListFilterParameters, Depends()],
+    coords: str | None = None,
+    name: str | None = None,
+    province: str | None = None,
+    region: str | None = None,
+):
+    """List Italian municipalities."""
+    geom_filter_kwarg = {}
+    if coords is not None:
+        geom = shapely.io.from_wkt(coords)
+        if geom.geom_type == "Point":
+            geom_filter_kwarg = {"point_filter": geom}
+        elif geom.geom_type == "Polygon":
+            geom_filter_kwarg = {"polygon_intersection_filter": geom}
+        else:
+            raise HTTPException(
+                status_code=400, detail="geometry be either point or polygon"
+            )
+    municipalities, filtered_total = db.list_municipalities(
+        db_session,
+        limit=list_params.limit,
+        offset=list_params.offset,
+        include_total=True,
+        name_filter=name,
+        province_name_filter=province,
+        region_name_filter=region,
+        **geom_filter_kwarg,
+    )
+    _, unfiltered_total = db.list_stations(
+        db_session, limit=1, offset=0, include_total=True
+    )
+    return municipalities_geojson.MunicipalityFeatureCollection.from_items(
+        municipalities,
+        request,
+        limit=list_params.limit,
+        offset=list_params.offset,
+        filtered_total=filtered_total,
+        unfiltered_total=unfiltered_total,
+    )

arpav_ppcv/webapp/api_v2/routers/observations.py

@@ -27,6 +27,7 @@
     database as db,
     operations,
 )
+from ...responses import GeoJsonResponse
 from ....schemas import base
 from ... import dependencies
 from ..schemas import observations
@@ -41,10 +42,6 @@
 router = APIRouter()
 
 
-class GeoJsonResponse(JSONResponse):
-    media_type = "application/geo+json"
-
-
 @router.get(
     "/stations",
     response_class=GeoJsonResponse,

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

@@ -0,0 +1,41 @@
+import geojson_pydantic
+import pydantic
+from fastapi import Request
+
+from .....schemas import (
+    fields,
+    municipalities,
+)
+from .base import ArpavFeatureCollection
+
+
+class MunicipalityFeatureCollectionItem(geojson_pydantic.Feature):
+    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+
+    type: str = "Feature"
+    id: pydantic.UUID4
+    geometry: fields.WkbElement
+
+    @classmethod
+    def from_db_instance(
+        cls,
+        instance: municipalities.Municipality,
+        request: Request,
+    ) -> "MunicipalityFeatureCollectionItem":
+        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

arpav_ppcv/webapp/responses.py

@@ -0,0 +1,5 @@
+from fastapi.responses import JSONResponse
+
+
+class GeoJsonResponse(JSONResponse):
+    media_type = "application/geo+json"