Pydantic input feature collection

ohsome_quality_api/api/api.py

@@ -13,7 +13,6 @@
     get_swagger_ui_oauth2_redirect_html,
 )
 from fastapi.responses import JSONResponse
-from geojson import FeatureCollection
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.staticfiles import StaticFiles
 
@@ -25,6 +24,7 @@
     oqt,
 )
 from ohsome_quality_api.api.request_models import (
+    FeatureCollection,
     IndicatorDataRequest,
     IndicatorRequest,
     ReportRequest,
@@ -247,7 +247,8 @@ async def post_indicator_ms(parameters: IndicatorDataRequest) -> CustomJSONRespo
         include_figure=parameters.include_figure,
     )
     geojson_object = FeatureCollection(
-        features=[i.as_feature(parameters.include_data) for i in indicators]
+        type="FeatureCollection",
+        features=[i.as_feature(parameters.include_data) for i in indicators],
     )
     response = empty_api_response()
     response["attribution"]["text"] = get_class_from_key(
@@ -332,7 +333,7 @@ async def post_report(
         class_type="report",
         key=key.value,
     ).attribution()
-    response.update(geojson_object)
+    response.update(geojson_object.model_dump())
     return CustomJSONResponse(content=response, media_type=MEDIA_TYPE_GEOJSON)
 
 

ohsome_quality_api/api/request_models.py

@@ -1,13 +1,23 @@
-import json
+# ruff: noqa: N805
+from typing import Any, Dict, Generic, Optional, TypeVar, Union
 
-import geojson
-from geojson import FeatureCollection
+from geojson_pydantic import (
+    Feature as PydanticFeature,
+)
+from geojson_pydantic import (
+    FeatureCollection as PydanticFeatureCollection,
+)
+from geojson_pydantic import (
+    MultiPolygon,
+    Polygon,
+)
+from geojson_pydantic.features import Feat, Geom, Props
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from ohsome_quality_api.topics.definitions import TopicEnum
 from ohsome_quality_api.topics.models import TopicData
+from ohsome_quality_api.utils.exceptions import InvalidCRSError
 from ohsome_quality_api.utils.helper import snake_to_lower_camel
-from ohsome_quality_api.utils.validators import validate_geojson
 
 
 class BaseConfig(BaseModel):
@@ -19,38 +29,38 @@ class BaseConfig(BaseModel):
     )
 
 
-class BaseBpolys(BaseConfig):
-    bpolys: dict = Field(
-        {
-            "type": "FeatureCollection",
-            "features": [
-                {
-                    "type": "Feature",
-                    "geometry": {
-                        "type": "Polygon",
-                        "coordinates": [
-                            [
-                                [8.674092292785645, 49.40427147224242],
-                                [8.695850372314453, 49.40427147224242],
-                                [8.695850372314453, 49.415552187316095],
-                                [8.674092292785645, 49.415552187316095],
-                                [8.674092292785645, 49.40427147224242],
-                            ]
-                        ],
-                    },
-                },
-            ],
-        }
-    )
+class Feature(PydanticFeature[Geom, Props], Generic[Geom, Props]):
+    """Extended Feature that make properties optional."""
+
+    properties: Optional[Union[Props, None]] = None
+
+
+Crs = TypeVar("Crs", bound=Union[Dict[str, Any], BaseModel])
+
 
-    @field_validator("bpolys")
-    @classmethod
-    def validate_bpolys(cls, value) -> FeatureCollection:
-        obj = geojson.loads(json.dumps(value))
-        if not isinstance(obj, FeatureCollection):
-            raise ValueError("must be of type FeatureCollection")
-        validate_geojson(obj)  # Check if exceptions are raised
-        return obj
+class FeatureCollection(PydanticFeatureCollection[Feat], Generic[Feat]):
+    """Extended FeatureCollection that also checks for CRS."""
+
+    crs: Optional[Union[Crs, None]] = None
+
+    @field_validator("crs", mode="before")
+    def check_crs(cls, crs: dict):
+        """Check if CRS is not WGS84"""
+        if crs is None:
+            return
+        if crs["type"] != "name":
+            raise InvalidCRSError(
+                "Invalid CRS. CRS object is not according to specification."
+            )
+        if crs["properties"]["name"] not in [
+            "urn\\:ogc:def:crs:OGC:1.3:CRS84",
+            "EPSG:4326",
+        ]:
+            raise InvalidCRSError("Invalid CRS. GeoJSON must be in WGS84 (EPSG:4326).")
+
+
+class BaseBpolys(BaseConfig):
+    bpolys: FeatureCollection[Feature[Polygon | MultiPolygon, Props]]
 
 
 class IndicatorRequest(BaseBpolys):

ohsome_quality_api/geodatabase/client.py

@@ -19,8 +19,8 @@
 
 import asyncpg
 from asyncpg import Record
-from geojson import Feature, FeatureCollection
 
+from ohsome_quality_api.api.request_models import Feature, FeatureCollection
 from ohsome_quality_api.config import get_config_value
 
 WORKING_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -43,7 +43,9 @@ async def get_connection():
         await conn.close()
 
 
-async def get_shdi(bpoly: Feature | FeatureCollection) -> list[Record]:
+async def get_shdi(
+    bpoly: Feature | FeatureCollection,
+) -> list[Record]:
     """Get Subnational Human Development Index (SHDI) for a bounding polygon.
 
     Get SHDI by intersecting the bounding polygon with sub-national regions provided by

ohsome_quality_api/indicators/attribute_completeness/indicator.py

@@ -4,8 +4,8 @@
 import dateutil.parser
 import numpy as np
 import plotly.graph_objects as go
-from geojson import Feature
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.attributes.definitions import get_attribute
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client

ohsome_quality_api/indicators/base.py

@@ -2,8 +2,8 @@
 from abc import ABCMeta, abstractmethod
 
 import plotly.graph_objects as go
-from geojson import Feature
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.definitions import get_attribution, get_metadata
 from ohsome_quality_api.indicators.models import IndicatorMetadata, Result
 from ohsome_quality_api.topics.models import BaseTopic as Topic
@@ -28,7 +28,7 @@ def __init__(
         )
         self._get_default_figure()
 
-    def as_dict(self, include_data: bool = False, exclude_label: bool = False) -> dict:
+    def as_dict(self, exclude_label: bool = False) -> dict:
         if exclude_label:
             result = self.result.model_dump(by_alias=True, exclude={"label"})
         else:
@@ -43,35 +43,27 @@ def as_dict(self, include_data: bool = False, exclude_label: bool = False) -> di
                 exclude={"ratio_filter"},
             ),
             "result": result,
-            **self.feature.properties,
         }
-        if include_data:
-            raw_dict["data"] = self.data
-        if "id" in self.feature.keys():
+        if self.feature.properties is not None:
+            raw_dict.update(self.feature.properties)
+        if self.feature.id is not None:
             raw_dict["id"] = self.feature.id
         return raw_dict
 
-    def as_feature(self, include_data: bool = False, exclude_label=False) -> Feature:
+    def as_feature(self, exclude_label=False) -> Feature:
         """Return a GeoJSON Feature object.
 
         The properties of the Feature contains the attributes of the indicator.
         The geometry (and properties) of the input GeoJSON object is preserved.
-
-        Args:
-            include_data (bool): If true include additional data in the properties.
         """
-        properties = self.as_dict(include_data, exclude_label)
-        if "id" in self.feature.keys():
-            return Feature(
-                id=self.feature.id,
-                geometry=self.feature.geometry,
-                properties=properties,
-            )
-        else:
-            return Feature(
-                geometry=self.feature.geometry,
-                properties=properties,
-            )
+        properties = self.as_dict(exclude_label)
+
+        return Feature(
+            type="Feature",
+            id=self.feature.id,
+            geometry=self.feature.geometry,
+            properties=properties,
+        )
 
     @property
     def data(self) -> dict:

ohsome_quality_api/indicators/building_completeness/indicator.py

@@ -3,13 +3,12 @@
 from string import Template
 
 import dateutil.parser
-import geojson
 import numpy as np
 import plotly.graph_objs as go
 from building_completeness_model import Predictor, Processor
-from geojson import Feature, FeatureCollection
 
 import ohsome_quality_api.geodatabase.client as db_client
+from ohsome_quality_api.api.request_models import Feature, FeatureCollection
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client
 from ohsome_quality_api.raster import client as raster_client
@@ -111,7 +110,7 @@ async def preprocess(self) -> None:
         self.covariates["ghs_pop"] = [i["sum"] or 0 for i in ghs_pop]
         self.covariates["ghs_pop_density"] = [
             pop / cell.properties["area"]
-            for pop, cell in zip(self.covariates["ghs_pop"], hex_cells["features"])
+            for pop, cell in zip(self.covariates["ghs_pop"], hex_cells.features)
         ]
         self.covariates.update(get_smod_class_share(hex_cells))
 
@@ -275,8 +274,8 @@ async def get_hex_cells(feature: Feature) -> FeatureCollection:
         query = file.read()
     async with db_client.get_connection() as conn:
         record = await conn.fetchrow(query, str(feature.geometry))
-    feature_collection = geojson.loads(record[0])
-    if feature_collection["features"] is None:
+    feature_collection = FeatureCollection[Feature](**record[0])
+    if feature_collection.features is None:
         raise HexCellsNotFoundError
     return feature_collection
 

ohsome_quality_api/indicators/currentness/indicator.py

@@ -16,9 +16,9 @@
 import plotly.graph_objects as pgo
 import yaml
 from dateutil.parser import isoparse
-from geojson import Feature
 from plotly.subplots import make_subplots
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.definitions import Color
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client

ohsome_quality_api/indicators/density/indicator.py

@@ -4,8 +4,8 @@
 import dateutil.parser
 import numpy as np
 import plotly.graph_objects as go
-from geojson import Feature
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client
 from ohsome_quality_api.topics.models import BaseTopic as Topic

ohsome_quality_api/indicators/mapping_saturation/indicator.py

@@ -4,9 +4,9 @@
 import numpy as np
 import plotly.graph_objects as pgo
 from dateutil.parser import isoparse
-from geojson import Feature
 from rpy2.rinterface_lib.embedded import RRuntimeError
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.definitions import Color
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.indicators.mapping_saturation import models

ohsome_quality_api/indicators/minimal/indicator.py

@@ -2,8 +2,8 @@
 from string import Template
 
 import dateutil.parser
-from geojson import Feature
 
+from ohsome_quality_api.api.request_models import Feature
 from ohsome_quality_api.indicators.base import BaseIndicator
 from ohsome_quality_api.ohsome import client as ohsome_client
 from ohsome_quality_api.topics.models import BaseTopic as Topic

ohsome_quality_api/ohsome/client.py

@@ -1,19 +1,13 @@
 import datetime
 import json
 from functools import singledispatch
+from json import JSONDecodeError
 
-import geojson
 import httpx
 from dateutil.parser import isoparse
-from geojson import Feature, FeatureCollection
 from schema import Or, Schema, SchemaError, Use
 
-# `geojson` uses `simplejson` if it is installed
-try:
-    from simplejson import JSONDecodeError
-except ImportError:
-    from json import JSONDecodeError
-
+from ohsome_quality_api.api.request_models import Feature, FeatureCollection
 from ohsome_quality_api.attributes.models import Attribute
 from ohsome_quality_api.config import get_config_value
 from ohsome_quality_api.topics.models import BaseTopic as Topic
@@ -115,7 +109,7 @@ async def query_ohsome_api(url: str, data: dict) -> dict:
             message = error.response.json()["error"]
         raise OhsomeApiError("Querying the ohsome API failed! " + message) from error
     try:
-        return geojson.loads(resp.content)
+        return json.loads(resp.content)
     except JSONDecodeError as error:
         raise OhsomeApiError(
             "Ohsome API returned invalid GeoJSON after streaming of the response. "
@@ -165,9 +159,12 @@ def build_data_dict(
     """
     data = {"filter": topic.filter}
     if isinstance(bpolys, Feature):
-        data["bpolys"] = json.dumps(FeatureCollection([bpolys]))
+        data["bpolys"] = FeatureCollection[Feature](
+            type="FeatureCollection",
+            features=[bpolys],
+        ).model_dump_json(exclude_none=True)
     elif isinstance(bpolys, FeatureCollection):
-        data["bpolys"] = json.dumps(bpolys)
+        data["bpolys"] = bpolys.model_dump_json(exclude_none=True)
     else:
         raise TypeError("Parameter 'bpolys' does not have expected type.")
     if time is not None: