Add full page for catalogs to support API functionality

Author: moradology

stac_fastapi/pgstac/stac_fastapi/pgstac/core.py

@@ -17,7 +17,7 @@
 from stac_fastapi.pgstac.types.search import PgstacSearch
 from stac_fastapi.types.core import AsyncBaseCoreClient
 from stac_fastapi.types.errors import InvalidQueryParameter, NotFoundError
-from stac_fastapi.types.hierarchy import browseable_catalog
+from stac_fastapi.types.hierarchy import browseable_catalog_page
 from stac_fastapi.types.stac import (
     Children,
     Collection,
@@ -117,9 +117,17 @@ async def get_root_children(self, **kwargs) -> Children:
         """
         request: Request = kwargs["request"]
         base_url = str(request.base_url)
+        extension_schemas = [
+            schema.schema_href for schema in self.extensions if schema.schema_href
+        ]
         catalog_children = [
-            browseable_catalog(child, base_url, child["catalog_id"]).dict(
-                exclude_unset=True
+            browseable_catalog_page(
+                child,
+                base_url,
+                child["catalog_id"],
+                self.stac_version,
+                self.conformance_classes(),
+                extension_schemas,
             )
             for child in self.hierarchy_definition["children"]
             if "catalog_id" in child

stac_fastapi/sqlalchemy/stac_fastapi/sqlalchemy/core.py

@@ -28,7 +28,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.hierarchy import browseable_catalog
+from stac_fastapi.types.hierarchy import browseable_catalog_page
 from stac_fastapi.types.search import BaseSearchPostRequest
 from stac_fastapi.types.stac import (
     Children,
@@ -115,10 +115,18 @@ def get_root_children(self, **kwargs) -> Children:
         """
         request: Request = kwargs["request"]
         base_url = str(request.base_url)
+        extension_schemas = [
+            schema.schema_href for schema in self.extensions if schema.schema_href
+        ]
         catalog_children = [
-            browseable_catalog(child, base_url, child["catalog_id"]).dict(
-                exclude_unset=True
-            )
+            browseable_catalog_page(
+                child,
+                base_url,
+                child["catalog_id"],
+                self.stac_version,
+                self.conformance_classes(),
+                extension_schemas,
+            ).dict(exclude_unset=True)
             for child in self.hierarchy_definition["children"]
             if "catalog_id" in child
         ]

stac_fastapi/testdata/joplin/hierarchy.json

@@ -13,6 +13,10 @@
                 ["joplin", "f7f164c9-cfdf-436d-a3f0-69864c38ba2a"],
                 ["joplin", "f734401c-2df0-4694-a353-cdd3ea760cdc"]
             ] 
+        }, {
+            "collection_id": "joplin",
+            "children": [],
+            "items": []
         }],
         "items": [
             ["joplin", "fe916452-ba6f-4631-9154-c249924a122d"],

stac_fastapi/types/stac_fastapi/types/core.py

@@ -20,8 +20,8 @@
 from stac_fastapi.types.extension import ApiExtension
 from stac_fastapi.types.hierarchy import (
     BrowseableNode,
-    browseable_catalog,
     browseable_catalog_link,
+    browseable_catalog_page,
     browseable_collection_link,
     browseable_item_link,
 )
@@ -364,16 +364,16 @@ def landing_page(self, **kwargs) -> stac_types.LandingPage:
         )
 
         # Add Collections links
-        collections = self.all_collections(request=kwargs["request"])
-        for collection in collections["collections"]:
-            landing_page["links"].append(
-                {
-                    "rel": Relations.child.value,
-                    "type": MimeTypes.json.value,
-                    "title": collection.get("title") or collection.get("id"),
-                    "href": urljoin(base_url, f"collections/{collection['id']}"),
-                }
-            )
+        # collections = self.all_collections(request=kwargs["request"])
+        # for collection in collections["collections"]:
+        #     landing_page["links"].append(
+        #         {
+        #             "rel": Relations.child.value,
+        #             "type": MimeTypes.json.value,
+        #             "title": collection.get("title") or collection.get("id"),
+        #             "href": urljoin(base_url, f"collections/{collection['id']}"),
+        #         }
+        #     )
 
         # Add links for browseable and children conformance
         if self.hierarchy_definition is not None:
@@ -550,8 +550,18 @@ def get_catalog(self, catalog_path: str, **kwargs) -> stac_types.Catalog:
                 for node in remaining_hierarchy["children"]
                 if node["catalog_id"] == fork
             )
-        return browseable_catalog(remaining_hierarchy, base_url, catalog_path).dict(
-            exclude_unset=True
+
+        extension_schemas = [
+            schema.schema_href for schema in self.extensions if schema.schema_href
+        ]
+
+        return browseable_catalog_page(
+            remaining_hierarchy,
+            base_url,
+            catalog_path,
+            self.stac_version,
+            self.conformance_classes(),
+            extension_schemas,
         )
 
     @abc.abstractmethod
@@ -625,17 +635,17 @@ async def landing_page(self, **kwargs) -> stac_types.LandingPage:
             extension_schemas=extension_schemas,
         )
 
-        # Add Collections links
-        collections = await self.all_collections(request=kwargs["request"])
-        for collection in collections["collections"]:
-            landing_page["links"].append(
-                {
-                    "rel": Relations.child.value,
-                    "type": MimeTypes.json.value,
-                    "title": collection.get("title") or collection.get("id"),
-                    "href": urljoin(base_url, f"collections/{collection['id']}"),
-                }
-            )
+        # # Add Collections links
+        # collections = await self.all_collections(request=kwargs["request"])
+        # for collection in collections["collections"]:
+        #     landing_page["links"].append(
+        #         {
+        #             "rel": Relations.child.value,
+        #             "type": MimeTypes.json.value,
+        #             "title": collection.get("title") or collection.get("id"),
+        #             "href": urljoin(base_url, f"collections/{collection['id']}"),
+        #         }
+        #     )
 
         # Add links for children and browseable conformance
         if self.hierarchy_definition is not None:
@@ -656,7 +666,9 @@ async def landing_page(self, **kwargs) -> stac_types.LandingPage:
                     )
                 if "catalog_id" in child:
                     landing_page["links"].append(
-                        browseable_catalog_link(child, base_url, child["catalog_id"])
+                        browseable_catalog_link(
+                            child, base_url, child["catalog_id"], ""
+                        )
                     )
             for item in self.hierarchy_definition["items"]:
                 landing_page["links"].append(browseable_item_link(item, base_url))
@@ -813,8 +825,18 @@ async def get_catalog(self, catalog_path: str, **kwargs) -> stac_types.Catalog:
                 for node in remaining_hierarchy["children"]
                 if node["catalog_id"] == fork
             )
-        return browseable_catalog(remaining_hierarchy, base_url, catalog_path).dict(
-            exclude_unset=True
+
+        extension_schemas = [
+            schema.schema_href for schema in self.extensions if schema.schema_href
+        ]
+
+        return browseable_catalog_page(
+            remaining_hierarchy,
+            base_url,
+            catalog_path,
+            self.stac_version,
+            self.conformance_classes(),
+            extension_schemas,
         )
 
     @abc.abstractmethod

stac_fastapi/types/stac_fastapi/types/hierarchy.py

@@ -9,13 +9,12 @@
 from typing import List, Optional, Tuple, TypedDict, Union
 from urllib.parse import urljoin
 
-from stac_pydantic import Catalog
 from stac_pydantic.links import Relations
 from stac_pydantic.shared import MimeTypes
-from stac_pydantic.version import STAC_VERSION
+
+from stac_fastapi.types import stac as stac_types
 
 ItemPath = Tuple[str, str]
-NodeType = str
 
 
 class BrowseableNode(TypedDict):
@@ -57,7 +56,7 @@ def browseable_collection_link(node: BrowseableNode, base_url: str) -> str:
         "rel": Relations.child.value,
         "type": MimeTypes.json,
         "title": node.get("title") or node.get("collection_id"),
-        "href": urljoin([base_url, f"collections/{node['collection_id']}"]),
+        "href": urljoin(base_url, f"collections/{node['collection_id']}"),
     }
 
 
@@ -70,22 +69,31 @@ def browseable_item_link(item_path: ItemPath, base_url: str):
     }
 
 
-def browseable_catalog(node: CatalogNode, base_url: str, catalog_path: str) -> Catalog:
-    """Generate a catalog based on a CatalogNode in a BrowseableNode tree."""
+def browseable_catalog_page(
+    catalog_node: CatalogNode,
+    base_url: str,
+    catalog_path: str,
+    stac_version: str,
+    conformance_classes: List[str],
+    extension_schemas: List[str],
+) -> stac_types.LandingPage:
+    """Generate a STAC API landing page/catalog."""
     catalog_links = [
         browseable_catalog_link(
             child, base_url, "/".join([catalog_path.strip("/"), child["catalog_id"]])
         )
-        for child in node["children"]
+        for child in catalog_node["children"]
         if "catalog_id" in child
     ]
     collection_links = [
         browseable_collection_link(child, base_url)
-        for child in node["children"]
+        for child in catalog_node["children"]
         if "collection_id" in child
     ]
     children_links = catalog_links + collection_links
-    item_links = [browseable_item_link(item, base_url) for item in node["items"]]
+    item_links = [
+        browseable_item_link(item, base_url) for item in catalog_node["items"]
+    ]
 
     split_catalog_path = catalog_path.split("/")
     if len(split_catalog_path) > 1:
@@ -96,26 +104,59 @@ def browseable_catalog(node: CatalogNode, base_url: str, catalog_path: str) -> C
         parent_href = base_url
 
     standard_links = [
+        {
+            "rel": Relations.self.value,
+            "type": MimeTypes.json,
+            "href": urljoin(base_url, f"/catalogs/{catalog_path.strip('/')}"),
+        },
         {
             "rel": Relations.root.value,
             "type": MimeTypes.json,
             "href": base_url,
         },
         {
-            "rel": Relations.self.value,
+            "rel": "data",
             "type": MimeTypes.json,
-            "href": urljoin(base_url, f"/catalogs/{catalog_path.strip('/')}"),
+            "href": urljoin(
+                base_url, f"/catalogs/{catalog_path.strip('/')}/collections"
+            ),
+        },
+        {
+            "rel": Relations.conformance.value,
+            "type": MimeTypes.json,
+            "title": "STAC/WFS3 conformance classes implemented by this api",
+            "href": urljoin(
+                base_url, f"/catalogs/{catalog_path.strip('/')}/conformance"
+            ),
+        },
+        {
+            "rel": Relations.search.value,
+            "type": MimeTypes.geojson,
+            "title": "STAC search",
+            "href": urljoin(base_url, f"/catalogs/{catalog_path.strip('/')}/search"),
+            "method": "GET",
+        },
+        {
+            "rel": Relations.search.value,
+            "type": MimeTypes.json,
+            "title": "STAC search",
+            "href": urljoin(base_url, f"/catalogs/{catalog_path.strip('/')}/search"),
+            "method": "POST",
         },
         {"rel": Relations.parent.value, "type": MimeTypes.json, "href": parent_href},
     ]
-    return Catalog(
+
+    catalog_page = stac_types.LandingPage(
         type="Catalog",
-        id=node["catalog_id"],
-        description=node.get("description")
-        or f"Generated description for {node['catalog_id']}",
-        stac_version=STAC_VERSION,
-        links=children_links + item_links + standard_links,
+        id=catalog_node["catalog_id"],
+        title=catalog_node["catalog_id"],
+        description=catalog_node["description"],
+        stac_version=stac_version,
+        conformsTo=conformance_classes,
+        links=standard_links + children_links + item_links,
+        stac_extensions=extension_schemas,
     )
+    return catalog_page
 
 
 def parse_hierarchy(d: dict) -> BrowseableNode: