Enable /catalogs route

compose.yml

@@ -23,6 +23,7 @@ services:
       - BACKEND=elasticsearch
       - DATABASE_REFRESH=true
       - ENABLE_COLLECTIONS_SEARCH_ROUTE=true
+      - ENABLE_CATALOG_ROUTE=true
       - REDIS_ENABLE=true
       - REDIS_HOST=redis
       - REDIS_PORT=6379
@@ -62,6 +63,7 @@ services:
       - BACKEND=opensearch
       - STAC_FASTAPI_RATE_LIMIT=200/minute
       - ENABLE_COLLECTIONS_SEARCH_ROUTE=true
+      - ENABLE_CATALOG_ROUTE=true
       - REDIS_ENABLE=true
       - REDIS_HOST=redis
       - REDIS_PORT=6379

stac_fastapi/core/stac_fastapi/core/extensions/__init__.py

@@ -1,5 +1,6 @@
 """elasticsearch extensions modifications."""
 
+from .catalogs import CatalogsExtension
 from .collections_search import CollectionsSearchEndpointExtension
 from .query import Operator, QueryableTypes, QueryExtension
 
@@ -8,4 +9,5 @@
     "QueryableTypes",
     "QueryExtension",
     "CollectionsSearchEndpointExtension",
+    "CatalogsExtension",
 ]

stac_fastapi/core/stac_fastapi/core/extensions/catalogs.py

@@ -0,0 +1,64 @@
+"""Catalogs extension."""
+
+from typing import List, Type, Union
+
+import attr
+from fastapi import APIRouter, FastAPI, Request
+from fastapi.responses import JSONResponse
+from starlette.responses import Response
+
+from stac_fastapi.types.core import BaseCoreClient
+from stac_fastapi.types.extension import ApiExtension
+from stac_fastapi.types.stac import LandingPage
+
+
+@attr.s
+class CatalogsExtension(ApiExtension):
+    """Catalogs Extension.
+
+    The Catalogs extension adds a /catalogs endpoint that returns the root catalog.
+    """
+
+    client: BaseCoreClient = attr.ib(default=None)
+    settings: dict = attr.ib(default=attr.Factory(dict))
+    conformance_classes: List[str] = attr.ib(default=attr.Factory(list))
+    router: APIRouter = attr.ib(default=attr.Factory(APIRouter))
+    response_class: Type[Response] = attr.ib(default=JSONResponse)
+
+    def register(self, app: FastAPI) -> None:
+        """Register the extension with a FastAPI application.
+
+        Args:
+            app: target FastAPI application.
+
+        Returns:
+            None
+        """
+        response_model = (
+            self.settings.get("response_model")
+            if isinstance(self.settings, dict)
+            else getattr(self.settings, "response_model", None)
+        )
+
+        self.router.add_api_route(
+            path="/catalogs",
+            endpoint=self.catalogs,
+            methods=["GET"],
+            response_model=LandingPage if response_model else None,
+            response_class=self.response_class,
+            summary="Get Catalogs",
+            description="Returns the root catalog.",
+            tags=["Catalogs"],
+        )
+        app.include_router(self.router, tags=["Catalogs"])
+
+    async def catalogs(self, request: Request) -> Union[LandingPage, Response]:
+        """Get catalogs.
+
+        Args:
+            request: Request object.
+
+        Returns:
+            The root catalog (landing page).
+        """
+        return await self.client.landing_page(request=request)

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py

@@ -23,6 +23,7 @@
     EsAggregationExtensionGetRequest,
     EsAggregationExtensionPostRequest,
 )
+from stac_fastapi.core.extensions.catalogs import CatalogsExtension
 from stac_fastapi.core.extensions.collections_search import (
     CollectionsSearchEndpointExtension,
 )
@@ -65,11 +66,13 @@
 ENABLE_COLLECTIONS_SEARCH_ROUTE = get_bool_env(
     "ENABLE_COLLECTIONS_SEARCH_ROUTE", default=False
 )
+ENABLE_CATALOG_ROUTE = get_bool_env("ENABLE_CATALOG_ROUTE", default=False)
 logger.info("TRANSACTIONS_EXTENSIONS is set to %s", TRANSACTIONS_EXTENSIONS)
 logger.info("ENABLE_COLLECTIONS_SEARCH is set to %s", ENABLE_COLLECTIONS_SEARCH)
 logger.info(
     "ENABLE_COLLECTIONS_SEARCH_ROUTE is set to %s", ENABLE_COLLECTIONS_SEARCH_ROUTE
 )
+logger.info("ENABLE_CATALOG_ROUTE is set to %s", ENABLE_CATALOG_ROUTE)
 
 settings = ElasticsearchSettings()
 session = Session.create_from_settings(settings)
@@ -202,6 +205,19 @@
     extensions.append(collections_search_endpoint_ext)
 
 
+if ENABLE_CATALOG_ROUTE:
+    catalogs_extension = CatalogsExtension(
+        client=CoreClient(
+            database=database_logic,
+            session=session,
+            post_request_model=collection_search_post_request_model,
+            landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
+        ),
+        settings=settings,
+    )
+    extensions.append(catalogs_extension)
+
+
 database_logic.extensions = [type(ext).__name__ for ext in extensions]
 
 post_request_model = create_post_request_model(search_extensions)

stac_fastapi/opensearch/stac_fastapi/opensearch/app.py

@@ -23,6 +23,7 @@
     EsAggregationExtensionGetRequest,
     EsAggregationExtensionPostRequest,
 )
+from stac_fastapi.core.extensions.catalogs import CatalogsExtension
 from stac_fastapi.core.extensions.collections_search import (
     CollectionsSearchEndpointExtension,
 )
@@ -65,11 +66,13 @@
 ENABLE_COLLECTIONS_SEARCH_ROUTE = get_bool_env(
     "ENABLE_COLLECTIONS_SEARCH_ROUTE", default=False
 )
+ENABLE_CATALOG_ROUTE = get_bool_env("ENABLE_CATALOG_ROUTE", default=False)
 logger.info("TRANSACTIONS_EXTENSIONS is set to %s", TRANSACTIONS_EXTENSIONS)
 logger.info("ENABLE_COLLECTIONS_SEARCH is set to %s", ENABLE_COLLECTIONS_SEARCH)
 logger.info(
     "ENABLE_COLLECTIONS_SEARCH_ROUTE is set to %s", ENABLE_COLLECTIONS_SEARCH_ROUTE
 )
+logger.info("ENABLE_CATALOG_ROUTE is set to %s", ENABLE_CATALOG_ROUTE)
 
 settings = OpensearchSettings()
 session = Session.create_from_settings(settings)
@@ -201,6 +204,19 @@
     extensions.append(collections_search_endpoint_ext)
 
 
+if ENABLE_CATALOG_ROUTE:
+    catalogs_extension = CatalogsExtension(
+        client=CoreClient(
+            database=database_logic,
+            session=session,
+            post_request_model=collection_search_post_request_model,
+            landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
+        ),
+        settings=settings,
+    )
+    extensions.append(catalogs_extension)
+
+
 database_logic.extensions = [type(ext).__name__ for ext in extensions]
 
 post_request_model = create_post_request_model(search_extensions)