Enable conformance class configuration

stac_fastapi/extensions/stac_fastapi/extensions/core/transaction.py

@@ -1,5 +1,5 @@
 """transaction extension."""
-from typing import Callable, Type, Union
+from typing import Callable, List, Type, Union
 
 import attr
 from fastapi import APIRouter, FastAPI
@@ -38,6 +38,7 @@ class TransactionExtension(ApiExtension):
     settings: ApiSettings = attr.ib()
     router: APIRouter = attr.ib(factory=APIRouter)
     response_class: Type[Response] = attr.ib(default=JSONResponse)
+    conformance_classes: List[str] = attr.ib(default=list())
 
     def _create_endpoint(
         self,

stac_fastapi/extensions/stac_fastapi/extensions/third_party/bulk_transactions.py

@@ -1,6 +1,6 @@
 """bulk transactions extension."""
 import abc
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 import attr
 from fastapi import APIRouter, FastAPI
@@ -60,6 +60,7 @@ class BulkTransactionExtension(ApiExtension):
     """
 
     client: BaseBulkTransactionsClient = attr.ib()
+    conformance_classes: List[str] = attr.ib(default=list())
 
     def register(self, app: FastAPI) -> None:
         """Register the extension with a FastAPI application.

stac_fastapi/extensions/stac_fastapi/extensions/third_party/tiles.py

@@ -171,6 +171,7 @@ class TilesExtension(ApiExtension):
 
     client: BaseTilesClient = attr.ib()
     route_prefix: str = attr.ib(default="/titiler")
+    conformance_classes: List[str] = attr.ib(default=list())
 
     def register(self, app: FastAPI) -> None:
         """Register the extension with a FastAPI application.

stac_fastapi/pgstac/stac_fastapi/pgstac/core.py

@@ -12,7 +12,7 @@
 from stac_fastapi.pgstac.types.search import PgstacSearch
 from stac_fastapi.types.core import AsyncBaseCoreClient
 from stac_fastapi.types.errors import NotFoundError
-from stac_fastapi.types.stac import Collection, Conformance, Item, ItemCollection
+from stac_fastapi.types.stac import Collection, Item, ItemCollection
 
 NumType = Union[float, int]
 
@@ -21,15 +21,6 @@
 class CoreCrudClient(AsyncBaseCoreClient):
     """Client for core endpoints defined by stac."""
 
-    async def conformance(self, **kwargs) -> Conformance:
-        """Conformance classes."""
-        return Conformance(
-            conformsTo=[
-                "https://stacspec.org/STAC-api.html",
-                "http://docs.opengeospatial.org/is/17-069r3/17-069r3.html#ats_geojson",
-            ]
-        )
-
     async def all_collections(self, **kwargs) -> List[Collection]:
         """Read all collections from the database."""
         request: Request = kwargs["request"]

stac_fastapi/sqlalchemy/stac_fastapi/sqlalchemy/core.py

@@ -26,7 +26,7 @@
 from stac_fastapi.types.config import Settings
 from stac_fastapi.types.core import BaseCoreClient
 from stac_fastapi.types.errors import NotFoundError
-from stac_fastapi.types.stac import Collection, Conformance, Item, ItemCollection
+from stac_fastapi.types.stac import Collection, Item, ItemCollection
 
 logger = logging.getLogger(__name__)
 
@@ -57,15 +57,6 @@ def _lookup_id(
             raise NotFoundError(f"{table.__name__} {id} not found")
         return row
 
-    def conformance(self, **kwargs) -> Conformance:
-        """Conformance classes."""
-        return Conformance(
-            conformsTo=[
-                "https://stacspec.org/STAC-api.html",
-                "http://docs.opengeospatial.org/is/17-069r3/17-069r3.html#ats_geojson",
-            ]
-        )
-
     def all_collections(self, **kwargs) -> List[Collection]:
         """Read all collections from the database."""
         base_url = str(kwargs["request"].base_url)

stac_fastapi/sqlalchemy/tests/resources/test_item.py

@@ -1,4 +1,5 @@
 import json
+import os
 import time
 import uuid
 from copy import deepcopy
@@ -11,6 +12,9 @@
 from shapely.geometry import Polygon
 from stac_pydantic.shared import DATETIME_RFC339
 
+from stac_fastapi.sqlalchemy.core import CoreCrudClient
+from stac_fastapi.types.core import LandingPageMixin
+
 
 def test_create_and_delete_item(app_client, load_test_data):
     """Test creation and deletion of a single item (transactions extension)"""
@@ -756,3 +760,18 @@ def test_search_invalid_query_field(app_client):
     body = {"query": {"gsd": {"lt": 100}, "invalid-field": {"eq": 50}}}
     resp = app_client.post("/search", json=body)
     assert resp.status_code == 400
+
+
+def test_conformance_classes_configurable():
+    """Test conformance class configurability"""
+    landing = LandingPageMixin()
+    landing_page = landing._landing_page(
+        base_url="http://test/test", conformance_classes=["this is a test"]
+    )
+    assert landing_page["conformsTo"][0] == "this is a test"
+
+    # Update environment to avoid key error on client instantiation
+    os.environ["READER_CONN_STRING"] = "testing"
+    os.environ["WRITER_CONN_STRING"] = "testing"
+    client = CoreCrudClient(base_conformance_classes=["this is a test"])
+    assert client.conformance_classes()[0] == "this is a test"

stac_fastapi/types/stac_fastapi/types/core.py

@@ -12,6 +12,7 @@
 
 from stac_fastapi.types import stac as stac_types
 from stac_fastapi.types.extension import ApiExtension
+from stac_fastapi.types.stac import Conformance
 
 NumType = Union[float, int]
 StacType = Dict[str, Any]
@@ -222,23 +223,24 @@ async def delete_collection(
 
 
 @attr.s
-class LandingPageMixin:
+class LandingPageMixin(abc.ABC):
     """Create a STAC landing page (GET /)."""
 
-    conformance_classes: List[str] = attr.ib()
     stac_version: str = attr.ib(default=STAC_VERSION)
     landing_page_id: str = attr.ib(default="stac-fastapi")
     title: str = attr.ib(default="stac-fastapi")
     description: str = attr.ib(default="stac-fastapi")
 
-    def _landing_page(self, base_url: str) -> stac_types.LandingPage:
+    def _landing_page(
+        self, base_url: str, conformance_classes: List[str]
+    ) -> stac_types.LandingPage:
         landing_page = stac_types.LandingPage(
             type="Catalog",
             id=self.landing_page_id,
             title=self.title,
             description=self.description,
             stac_version=self.stac_version,
-            conformsTo=self.conformance_classes,
+            conformsTo=conformance_classes,
             links=[
                 {
                     "rel": Relations.self.value,
@@ -282,13 +284,27 @@ class BaseCoreClient(LandingPageMixin, abc.ABC):
         extensions: list of registered api extensions.
     """
 
-    extensions: List[ApiExtension] = attr.ib(default=attr.Factory(list))
-    conformance_classes: List[str] = attr.ib(
+    base_conformance_classes: List[str] = attr.ib(
         factory=lambda: [
-            "https://stacspec.org/STAC-api.html",
-            "http://docs.opengeospatial.org/is/17-069r3/17-069r3.html#ats_geojson",
+            "https://api.stacspec.org/v1.0.0-beta.2/core",
+            "https://api.stacspec.org/v1.0.0-beta.2/ogcapi-features",
+            "https://api.stacspec.org/v1.0.0-beta.2/item-search",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/core",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/oas30",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/geojson",
         ]
     )
+    extensions: List[ApiExtension] = attr.ib(default=attr.Factory(list))
+
+    def conformance_classes(self) -> List[str]:
+        """Generate conformance classes by adding extension conformance to base conformance classes."""
+        base_conformance_classes = self.base_conformance_classes.copy()
+
+        for extension in self.extensions:
+            extension_classes = getattr(extension, "conformance_classes", [])
+            base_conformance_classes.extend(extension_classes)
+
+        return list(set(base_conformance_classes))
 
     def extension_is_enabled(self, extension: Type[ApiExtension]) -> bool:
         """Check if an api extension is enabled."""
@@ -303,7 +319,9 @@ def landing_page(self, **kwargs) -> stac_types.LandingPage:
             API landing page, serving as an entry point to the API.
         """
         base_url = str(kwargs["request"].base_url)
-        landing_page = self._landing_page(base_url=base_url)
+        landing_page = self._landing_page(
+            base_url=base_url, conformance_classes=self.conformance_classes()
+        )
         collections = self.all_collections(request=kwargs["request"])
         for collection in collections:
             landing_page["links"].append(
@@ -324,7 +342,7 @@ def conformance(self, **kwargs) -> stac_types.Conformance:
         Returns:
             Conformance classes which the server conforms to.
         """
-        stac_types.Conformance(conformsTo=self.conformance_classes)
+        return Conformance(conformsTo=self.conformance_classes())
 
     @abc.abstractmethod
     def post_search(
@@ -432,13 +450,27 @@ class AsyncBaseCoreClient(LandingPageMixin, abc.ABC):
         extensions: list of registered api extensions.
     """
 
-    extensions: List[ApiExtension] = attr.ib(default=attr.Factory(list))
-    conformance_classes: List[str] = attr.ib(
+    base_conformance_classes: List[str] = attr.ib(
         factory=lambda: [
-            "https://stacspec.org/STAC-api.html",
-            "http://docs.opengeospatial.org/is/17-069r3/17-069r3.html#ats_geojson",
+            "https://api.stacspec.org/v1.0.0-beta.2/core",
+            "https://api.stacspec.org/v1.0.0-beta.2/ogcapi-features",
+            "https://api.stacspec.org/v1.0.0-beta.2/item-search",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/core",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/oas30",
+            "http://www.opengis.net/spec/ogcapi-features-1/1.0/conf/geojson",
         ]
     )
+    extensions: List[ApiExtension] = attr.ib(default=attr.Factory(list))
+
+    def conformance_classes(self) -> List[str]:
+        """Generate conformance classes by adding extension conformance to base conformance classes."""
+        conformance_classes = self.base_conformance_classes.copy()
+
+        for extension in self.extensions:
+            extension_classes = getattr(extension, "conformance_classes", [])
+            conformance_classes.extend(extension_classes)
+
+        return list(set(conformance_classes))
 
     def extension_is_enabled(self, extension: Type[ApiExtension]) -> bool:
         """Check if an api extension is enabled."""
@@ -453,7 +485,9 @@ async def landing_page(self, **kwargs) -> stac_types.LandingPage:
             API landing page, serving as an entry point to the API.
         """
         base_url = str(kwargs["request"].base_url)
-        landing_page = self._landing_page(base_url=base_url)
+        landing_page = self._landing_page(
+            base_url=base_url, conformance_classes=self.conformance_classes()
+        )
         collections = await self.all_collections(request=kwargs["request"])
         for collection in collections:
             landing_page["links"].append(
@@ -474,7 +508,7 @@ async def conformance(self, **kwargs) -> stac_types.Conformance:
         Returns:
             Conformance classes which the server conforms to.
         """
-        stac_types.Conformance(conformsTo=self.conformance_classes)
+        return Conformance(conformsTo=self.conformance_classes())
 
     @abc.abstractmethod
     async def post_search(

stac_fastapi/types/stac_fastapi/types/extension.py

@@ -1,5 +1,6 @@
 """base api extension."""
 import abc
+from typing import List
 
 import attr
 from fastapi import FastAPI
@@ -9,6 +10,8 @@
 class ApiExtension(abc.ABC):
     """Abstract base class for defining API extensions."""
 
+    conformance_classes: List[str] = attr.ib(factory=list)
+
     @abc.abstractmethod
     def register(self, app: FastAPI) -> None:
         """Register the extension with a FastAPI application.