Add stac-pydantic support for validation

src/authorizer.py

@@ -1,4 +1,5 @@
 import json
+
 from fastapi import Request
 from globus_sdk import ConfidentialAppAuthClient, AccessTokenAuthorizer, GroupsClient
 from globus_sdk.scopes import GroupsScopes

src/client.py

@@ -1,6 +1,9 @@
 import json
+
 from datetime import datetime
 from fastapi import HTTPException, Request, Response, status
+from pydantic import ValidationError
+from settings.config.CMIP6ItemModel import CMIP6Item
 from stac_fastapi.types.core import BaseTransactionsClient
 from stac_fastapi.types.core import Collection, Item
 from typing import Optional, Union
@@ -109,6 +112,13 @@ async def create_item(
         auth = self.authorize(item, request, collection_id)
         user_agent = request.headers.get("headers", {}).get("User-Agent", "/").split("/")
 
+        stac_item = await request.json()
+        try:
+            CMIP6Item(**stac_item)
+        except ValidationError as e:
+            print(e.errors())
+            raise HTTPException(status_code=400, detail=str(e.errors()))
+
         message = {
             "metadata": {
                 "auth": auth,
@@ -125,7 +135,7 @@ async def create_item(
                 "payload": {
                     "method": "POST",
                     "collection_id": collection_id,
-                    "item": await request.json(),
+                    "item": stac_item,
                 },
             },
         }

src/producer.py

@@ -1,5 +1,6 @@
-from abc import ABC, abstractmethod
 import attr
+
+from abc import ABC, abstractmethod
 from confluent_kafka import Producer
 
 

src/settings/config/CMIP6ItemModel.py

@@ -0,0 +1,50 @@
+from pydantic import ConfigDict, HttpUrl
+from stac_pydantic import Item, ItemProperties
+from typing import List, Optional
+
+import datetime as datetimevalidate
+
+
+class CMIP6ItemProperties(ItemProperties):
+    access: List[str]
+    activity_id: List[str]
+    cf_standard_name: str
+    citation_url: HttpUrl
+    data_spec_version: Optional[str] = None
+    datetime: Optional[datetimevalidate.datetime] = None
+    end_datetime: datetimevalidate.datetime
+    experiment_id: str
+    experiment_title: str
+    frequency: str
+    further_info_url: HttpUrl
+    grid: str
+    grid_label: str
+    institution_id: str
+    mip_era: str
+    model_cohort: str
+    nominal_resolution: str
+    pid: str
+    product: str
+    project: str
+    realm: List[str]
+    retracted: Optional[str] = None
+    source_id: str
+    source_type: List[str]
+    start_datetime: datetimevalidate.datetime
+    sub_experiment_id: str
+    table_id: str
+    title: str
+    variable: str
+    variable_id: str
+    variable_long_name: str
+    variable_units: str
+    variant_label: str
+    version: str
+
+    model_config = ConfigDict(
+        protected_namespaces=()
+    )
+
+
+class CMIP6Item(Item):
+    properties: CMIP6ItemProperties