Merge pull request #67 from ricardogsilva/66-return-stations-list-as-…

…a-geojson-feature-collection-

Return GeoJSON FeatureCollection from stations list API endpoint

arpav_ppcv/schemas/fields.py

@@ -0,0 +1,16 @@
+import json
+from typing import Annotated
+
+import geoalchemy2
+import shapely.io
+from pydantic.functional_serializers import PlainSerializer
+
+def serialize_wkbelement(wkbelement: geoalchemy2.WKBElement):
+    geom = shapely.io.from_wkb(bytes(wkbelement.data))
+    return json.loads(shapely.io.to_geojson(geom))
+
+
+WkbElement = Annotated[
+    geoalchemy2.WKBElement,
+    PlainSerializer(serialize_wkbelement, return_type=dict, when_used="json")
+]

arpav_ppcv/schemas/models.py

@@ -1,29 +1,14 @@
 import datetime as dt
-import json
 import uuid
-from typing import (
-    Annotated,
-    Optional,
-)
+from typing import Optional
 
 import geojson_pydantic
 import geoalchemy2
 import pydantic
 import sqlalchemy
-import shapely.io
 import sqlmodel
-from pydantic.functional_serializers import PlainSerializer
 
-
-def serialize_wkbelement(wkbelement: geoalchemy2.WKBElement):
-    geom = shapely.io.from_wkb(bytes(wkbelement.data))
-    return json.loads(shapely.io.to_geojson(geom))
-
-
-WkbElement = Annotated[
-    geoalchemy2.WKBElement,
-    PlainSerializer(serialize_wkbelement, return_type=dict, when_used="json")
-]
+from . import fields
 
 
 class StationBase(sqlmodel.SQLModel):
@@ -33,7 +18,7 @@ class StationBase(sqlmodel.SQLModel):
         default_factory=uuid.uuid4,
         primary_key=True
     )
-    geom: WkbElement = sqlmodel.Field(
+    geom: fields.WkbElement = sqlmodel.Field(
         sa_column=sqlalchemy.Column(
             geoalchemy2.Geometry(
                 srid=4326,

arpav_ppcv/webapp/api_v2/routers/observations.py

@@ -5,23 +5,45 @@
 from fastapi import (
     APIRouter,
     Depends,
+    Header,
     Request,
 )
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
 from sqlmodel import Session
 
 from .... import database
 from ... import dependencies
 from ..schemas import observations
+from ..schemas.geojson import observations as observations_geojson
 
 logger = logging.getLogger(__name__)
 router = APIRouter()
 
 
-@router.get("/stations", response_model=observations.StationList)
+class GeoJsonResponse(JSONResponse):
+    media_type = "application/geo+json"
+
+
+@router.get(
+    "/stations",
+    response_class=GeoJsonResponse,
+    response_model=observations_geojson.StationFeatureCollection,
+    responses={
+        200: {
+            "content": {"application/json": {}},
+            "description": (
+                "Return a GeoJSON feature collection or a custom JSON "
+                "representation of the stations"
+            )
+        }
+    }
+)
 def list_stations(
         request: Request,
         db_session: Annotated[Session, Depends(dependencies.get_db_session)],
-        list_params: Annotated[dependencies.CommonListFilterParameters, Depends()]
+        list_params: Annotated[dependencies.CommonListFilterParameters, Depends()],
+        accept: Annotated[str | None, Header()] = None
 ):
     """List known stations."""
     stations, filtered_total = database.list_stations(
@@ -33,14 +55,29 @@ def list_stations(
     _, unfiltered_total = database.list_stations(
         db_session, limit=1, offset=0, include_total=True
     )
-    return observations.StationList.from_items(
-        stations,
-        request,
-        limit=list_params.limit,
-        offset=list_params.offset,
-        filtered_total=filtered_total,
-        unfiltered_total=unfiltered_total
-    )
+    if accept == "application/json":
+        result = JSONResponse(
+            content=jsonable_encoder(
+                observations.StationList.from_items(
+                    stations,
+                    request,
+                    limit=list_params.limit,
+                    offset=list_params.offset,
+                    filtered_total=filtered_total,
+                    unfiltered_total=unfiltered_total
+                )
+            )
+        )
+    else:
+        result = observations_geojson.StationFeatureCollection.from_items(
+            stations,
+            request,
+            limit=list_params.limit,
+            offset=list_params.offset,
+            filtered_total=filtered_total,
+            unfiltered_total=unfiltered_total
+        )
+    return result
 
 
 @router.get(

arpav_ppcv/webapp/api_v2/schemas/base.py

@@ -83,7 +83,7 @@ def _get_list_links(
             del filters["limit"]
         if "offset" in filters.keys():
             del filters["offset"]
-        pagination_urls = _get_pagination_urls(
+        pagination_urls = get_pagination_urls(
             request.url_for(cls.path_operation_name),
             num_returned_records,
             filtered_total,
@@ -104,7 +104,7 @@ def _get_meta(
         )
 
 
-def _get_pagination_urls(
+def get_pagination_urls(
         base_url: str,
         returned_records: int,
         total_records: int,

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

@@ -0,0 +1,85 @@
+import typing
+
+import geojson_pydantic
+import pydantic
+from fastapi.requests import Request
+
+from ..base import (
+    ListLinks,
+    get_pagination_urls,
+)
+
+F = typing.TypeVar("F", bound="ApiReadableModelAsFeature")
+
+
+@typing.runtime_checkable
+class ApiReadableModelAsFeature(typing.Protocol):
+    """Protocol to be used by all schema models that represent API resources.
+
+    It includes the `from_db_instance()` class method, which is to be used for
+    constructing instances.
+    """
+
+    @classmethod
+    def from_db_instance(  # noqa: D102
+            cls: typing.Type[F], db_instance: pydantic.BaseModel, request: Request
+    ) -> F:
+        ...
+
+
+class ArpavFeatureCollection(geojson_pydantic.FeatureCollection):
+    list_item_type: typing.ClassVar[typing.Type[ApiReadableModelAsFeature]]
+    path_operation_name: typing.ClassVar[str]
+
+    type: str = "FeatureCollection"
+    links: ListLinks
+    number_matched: int
+    number_total: int
+    number_returned: int
+
+    @classmethod
+    def from_items(
+            cls,
+            items: typing.Sequence[pydantic.BaseModel],
+            request: Request,
+            *,
+            limit: int,
+            offset: int,
+            filtered_total: int,
+            unfiltered_total: int
+    ) -> "cls":
+        return cls(
+            features=[
+                cls.list_item_type.from_db_instance(i, request)
+                for i in items
+            ],
+            links=cls._get_list_links(
+                request, limit, offset, filtered_total, len(items)),
+            number_matched=filtered_total,
+            number_total=unfiltered_total,
+            number_returned=len(items),
+        )
+
+    @classmethod
+    def _get_list_links(
+            cls,
+            request: Request,
+            limit: int,
+            offset: int,
+            filtered_total: int,
+            num_returned_records: int
+    ) -> ListLinks:
+        filters = dict(request.query_params)
+        if "limit" in filters.keys():
+            del filters["limit"]
+        if "offset" in filters.keys():
+            del filters["offset"]
+        pagination_urls = get_pagination_urls(
+            request.url_for(cls.path_operation_name),
+            num_returned_records,
+            filtered_total,
+            limit,
+            offset,
+            **filters,
+        )
+        return ListLinks(**pagination_urls)

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

@@ -0,0 +1,42 @@
+from fastapi import Request
+import geojson_pydantic
+import pydantic
+
+from .....schemas import (
+    models as app_models,
+    fields,
+)
+from .base import ArpavFeatureCollection
+
+
+class StationFeatureCollectionItem(geojson_pydantic.Feature):
+    model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
+
+    type: str = "Feature"
+    id: pydantic.UUID4
+    geometry: fields.WkbElement
+    links: list[str]
+
+    @classmethod
+    def from_db_instance(
+            cls,
+            instance: app_models.Station,
+            request: Request,
+    ) -> "StationFeatureCollectionItem":
+        url = request.url_for("get_station", **{"station_id": instance.id})
+        return cls(
+            id=instance.id,
+            geometry=instance.geom,
+            properties=instance.model_dump(
+                exclude={
+                    "id",
+                    "geom",
+                }
+            ),
+            links=[str(url)]
+        )
+
+
+class StationFeatureCollection(ArpavFeatureCollection):
+    path_operation_name = "list_stations"
+    list_item_type = StationFeatureCollectionItem

arpav_ppcv/webapp/api_v2/schemas/observations.py

@@ -1,4 +1,5 @@
 import logging
+
 import pydantic
 from fastapi import Request
 

tests/test_webapp_v2_routers_observations.py

@@ -10,11 +10,23 @@ def test_station_list(
         sample_stations: list[models.Station]
 ):
     list_response = test_client_v2_app.get(
-        test_client_v2_app.app.url_path_for("list_stations"))
+        test_client_v2_app.app.url_path_for("list_stations"),
+        headers={"accept": "application/json"}
+    )
     assert list_response.status_code == 200
     assert len(list_response.json()["items"]) == 20
 
 
+def test_station_list_geojson(
+        test_client_v2_app: httpx.Client,
+        sample_stations: list[models.Station]
+):
+    list_response = test_client_v2_app.get(
+        test_client_v2_app.app.url_path_for("list_stations"))
+    assert list_response.status_code == 200
+    assert len(list_response.json()["features"]) == 20
+
+
 def test_station_detail(
         test_client_v2_app: httpx.Client,
         sample_stations: list[models.Station]