add item collections routes etc

Author: jonhealy1

stac_fastapi/core/stac_fastapi/core/core.py

@@ -23,7 +23,6 @@
 from stac_fastapi.core.base_database_logic import BaseDatabaseLogic
 from stac_fastapi.core.base_settings import ApiBaseSettings
 from stac_fastapi.core.datetime_utils import format_datetime_range
-from stac_fastapi.core.models import Catalog
 from stac_fastapi.core.models.links import PagingLinks
 from stac_fastapi.core.redis_utils import redis_pagination_links
 from stac_fastapi.core.serializers import (
@@ -94,6 +93,17 @@ class CoreClient(AsyncBaseCoreClient):
     title: str = attr.ib(default="stac-fastapi")
     description: str = attr.ib(default="stac-fastapi")
 
+    def extension_is_enabled(self, extension_name: str) -> bool:
+        """Check if an extension is enabled by checking self.extensions.
+
+        Args:
+            extension_name: Name of the extension class to check for.
+
+        Returns:
+            True if the extension is in self.extensions, False otherwise.
+        """
+        return any(ext.__class__.__name__ == extension_name for ext in self.extensions)
+
     def _landing_page(
         self,
         base_url: str,
@@ -148,24 +158,16 @@ def _landing_page(
         )
         return landing_page
 
-    async def landing_page(self, **kwargs) -> Union[stac_types.LandingPage, Catalog]:
+    async def landing_page(self, **kwargs) -> stac_types.LandingPage:
         """Landing page.
 
         Called with `GET /`.
 
         Returns:
-            API landing page, serving as an entry point to the API, or root catalog if CatalogsExtension is enabled.
+            API landing page, serving as an entry point to the API.
         """
         request: Request = kwargs["request"]
 
-        # If CatalogsExtension is enabled, return root catalog instead of landing page
-        if self.extension_is_enabled("CatalogsExtension"):
-            # Find the CatalogsExtension and call its catalogs method
-            for extension in self.extensions:
-                if extension.__class__.__name__ == "CatalogsExtension":
-                    return await extension.catalogs(request)
-
-        # Normal landing page logic
         base_url = get_base_url(request)
         landing_page = self._landing_page(
             base_url=base_url,
@@ -223,6 +225,16 @@ async def landing_page(self, **kwargs) -> Union[stac_types.LandingPage, Catalog]
                 ]
             )
 
+        if self.extension_is_enabled("CatalogsExtension"):
+            landing_page["links"].append(
+                {
+                    "rel": "catalogs",
+                    "type": "application/json",
+                    "title": "Catalogs",
+                    "href": urljoin(base_url, "catalogs"),
+                }
+            )
+
         # Add OpenAPI URL
         landing_page["links"].append(
             {

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

@@ -84,6 +84,42 @@ def register(self, app: FastAPI, settings=None) -> None:
             tags=["Catalogs"],
         )
 
+        # Add endpoint for getting a specific collection in a catalog
+        self.router.add_api_route(
+            path="/catalogs/{catalog_id}/collections/{collection_id}",
+            endpoint=self.get_catalog_collection,
+            methods=["GET"],
+            response_model=stac_types.Collection,
+            response_class=self.response_class,
+            summary="Get Catalog Collection",
+            description="Get a specific collection from a catalog.",
+            tags=["Catalogs"],
+        )
+
+        # Add endpoint for getting items in a collection within a catalog
+        self.router.add_api_route(
+            path="/catalogs/{catalog_id}/collections/{collection_id}/items",
+            endpoint=self.get_catalog_collection_items,
+            methods=["GET"],
+            response_model=stac_types.ItemCollection,
+            response_class=self.response_class,
+            summary="Get Catalog Collection Items",
+            description="Get items from a collection in a catalog.",
+            tags=["Catalogs"],
+        )
+
+        # Add endpoint for getting a specific item in a collection within a catalog
+        self.router.add_api_route(
+            path="/catalogs/{catalog_id}/collections/{collection_id}/items/{item_id}",
+            endpoint=self.get_catalog_collection_item,
+            methods=["GET"],
+            response_model=stac_types.Item,
+            response_class=self.response_class,
+            summary="Get Catalog Collection Item",
+            description="Get a specific item from a collection in a catalog.",
+            tags=["Catalogs"],
+        )
+
         app.include_router(self.router, tags=["Catalogs"])
 
     async def catalogs(self, request: Request) -> Catalog:
@@ -253,3 +289,82 @@ async def get_catalog_collections(
             raise HTTPException(
                 status_code=404, detail=f"Catalog {catalog_id} not found"
             )
+
+    async def get_catalog_collection(
+        self, catalog_id: str, collection_id: str, request: Request
+    ) -> stac_types.Collection:
+        """Get a specific collection from a catalog.
+
+        Args:
+            catalog_id: The ID of the catalog.
+            collection_id: The ID of the collection.
+            request: Request object.
+
+        Returns:
+            The requested collection.
+        """
+        # Verify the catalog exists
+        try:
+            await self.client.database.find_catalog(catalog_id)
+        except Exception:
+            raise HTTPException(
+                status_code=404, detail=f"Catalog {catalog_id} not found"
+            )
+
+        # Delegate to the core client's get_collection method
+        return await self.client.get_collection(
+            collection_id=collection_id, request=request
+        )
+
+    async def get_catalog_collection_items(
+        self, catalog_id: str, collection_id: str, request: Request
+    ) -> stac_types.ItemCollection:
+        """Get items from a collection in a catalog.
+
+        Args:
+            catalog_id: The ID of the catalog.
+            collection_id: The ID of the collection.
+            request: Request object.
+
+        Returns:
+            ItemCollection containing items from the collection.
+        """
+        # Verify the catalog exists
+        try:
+            await self.client.database.find_catalog(catalog_id)
+        except Exception:
+            raise HTTPException(
+                status_code=404, detail=f"Catalog {catalog_id} not found"
+            )
+
+        # Delegate to the core client's item_collection method
+        return await self.client.item_collection(
+            collection_id=collection_id, request=request
+        )
+
+    async def get_catalog_collection_item(
+        self, catalog_id: str, collection_id: str, item_id: str, request: Request
+    ) -> stac_types.Item:
+        """Get a specific item from a collection in a catalog.
+
+        Args:
+            catalog_id: The ID of the catalog.
+            collection_id: The ID of the collection.
+            item_id: The ID of the item.
+            request: Request object.
+
+        Returns:
+            The requested item.
+        """
+        # Verify the catalog exists
+        try:
+            await self.client.database.find_catalog(catalog_id)
+        except Exception:
+            raise HTTPException(
+                status_code=404, detail=f"Catalog {catalog_id} not found"
+            )
+
+        # Delegate to the core client's get_item method
+        return await self.client.get_item(
+            item_id=item_id, collection_id=collection_id, request=request
+        )

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/app.py

@@ -214,6 +214,9 @@
             landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
         ),
         settings=settings,
+        conformance_classes=[
+            "https://api.stacspec.org/v1.0.0-beta.1/catalogs",
+        ],
     )
     extensions.append(catalogs_extension)
 

stac_fastapi/opensearch/stac_fastapi/opensearch/app.py

@@ -213,6 +213,9 @@
             landing_page_id=os.getenv("STAC_FASTAPI_LANDING_PAGE_ID", "stac-fastapi"),
         ),
         settings=settings,
+        conformance_classes=[
+            "https://api.stacspec.org/v1.0.0-beta.1/catalogs",
+        ],
     )
     extensions.append(catalogs_extension)