return list of catalogs

Author: jonhealy1

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

@@ -2,12 +2,13 @@
 
 import logging
 from typing import List, Optional, Type
-from urllib.parse import urlparse
+from urllib.parse import urlencode, urlparse
 
 import attr
 from fastapi import APIRouter, FastAPI, HTTPException, Query, Request
 from fastapi.responses import JSONResponse
 from starlette.responses import Response
+from typing_extensions import TypedDict
 
 from stac_fastapi.core.models import Catalog
 from stac_fastapi.types import stac as stac_types
@@ -17,12 +18,24 @@
 logger = logging.getLogger(__name__)
 
 
+class Catalogs(TypedDict, total=False):
+    """Catalogs endpoint response.
+
+    Similar to Collections but for catalogs.
+    """
+
+    catalogs: List[Catalog]
+    links: List[dict]
+    numberMatched: Optional[int]
+    numberReturned: Optional[int]
+
+
 @attr.s
 class CatalogsExtension(ApiExtension):
     """Catalogs Extension.
 
-    The Catalogs extension adds a /catalogs endpoint that returns the root catalog
-    containing child links to all catalogs in the database.
+    The Catalogs extension adds a /catalogs endpoint that returns a list of all catalogs
+    in the database, similar to how /collections returns a list of collections.
     """
 
     client: BaseCoreClient = attr.ib(default=None)
@@ -44,10 +57,10 @@ def register(self, app: FastAPI, settings=None) -> None:
             path="/catalogs",
             endpoint=self.catalogs,
             methods=["GET"],
-            response_model=Catalog,
+            response_model=Catalogs,
             response_class=self.response_class,
-            summary="Get Root Catalog",
-            description="Returns the root catalog containing links to all catalogs.",
+            summary="Get All Catalogs",
+            description="Returns a list of all catalogs in the database.",
             tags=["Catalogs"],
         )
 
@@ -140,73 +153,57 @@ async def catalogs(
             None,
             description="Pagination token for the next page of results",
         ),
-    ) -> Catalog:
-        """Get root catalog with links to all catalogs.
+    ) -> Catalogs:
+        """Get all catalogs with pagination support.
 
         Args:
             request: Request object.
             limit: The maximum number of catalogs to return (page size). Defaults to 10.
             token: Pagination token for the next page of results.
 
         Returns:
-            Root catalog containing child links to all catalogs in the database.
+            Catalogs object containing catalogs and pagination links.
         """
         base_url = str(request.base_url)
 
         # Get all catalogs from database with pagination
-        catalogs, _, _ = await self.client.database.get_all_catalogs(
+        catalogs, next_token, _ = await self.client.database.get_all_catalogs(
             token=token,
             limit=limit,
             request=request,
             sort=[{"field": "id", "direction": "asc"}],
         )
 
-        # Create child links to each catalog
-        child_links = []
+        # Convert database catalogs to STAC format
+        catalog_stac_objects = []
         for catalog in catalogs:
-            catalog_id = catalog.get("id") if isinstance(catalog, dict) else catalog.id
-            catalog_title = (
-                catalog.get("title") or catalog_id
-                if isinstance(catalog, dict)
-                else catalog.title or catalog.id
-            )
-            child_links.append(
-                {
-                    "rel": "child",
-                    "href": f"{base_url}catalogs/{catalog_id}",
-                    "type": "application/json",
-                    "title": catalog_title,
-                }
-            )
-
-        # Create root catalog
-        root_catalog = {
-            "type": "Catalog",
-            "stac_version": "1.0.0",
-            "id": "root",
-            "title": "Root Catalog",
-            "description": "Root catalog containing all available catalogs",
-            "links": [
-                {
-                    "rel": "self",
-                    "href": f"{base_url}catalogs",
-                    "type": "application/json",
-                },
-                {
-                    "rel": "root",
-                    "href": f"{base_url}catalogs",
-                    "type": "application/json",
-                },
-                {
-                    "rel": "parent",
-                    "href": base_url.rstrip("/"),
-                    "type": "application/json",
-                },
-            ]
-            + child_links,
-        }
-
-        return Catalog(**root_catalog)
+            catalog_stac = self.client.catalog_serializer.db_to_stac(catalog, request)
+            catalog_stac_objects.append(catalog_stac)
+
+        # Create pagination links
+        links = [
+            {"rel": "root", "type": "application/json", "href": base_url},
+            {"rel": "parent", "type": "application/json", "href": base_url},
+            {"rel": "self", "type": "application/json", "href": str(request.url)},
+        ]
+
+        # Add next link if there are more pages
+        if next_token:
+            query_params = {"limit": limit, "token": next_token}
+            next_link = {
+                "rel": "next",
+                "href": f"{base_url}catalogs?{urlencode(query_params)}",
+                "type": "application/json",
+                "title": "Next page of catalogs",
+            }
+            links.append(next_link)
+
+        # Return Catalogs object with catalogs
+        return Catalogs(
+            catalogs=catalog_stac_objects,
+            links=links,
+            numberReturned=len(catalog_stac_objects),
+        )
 
     async def create_catalog(self, catalog: Catalog, request: Request) -> Catalog:
         """Create a new catalog.

stac_fastapi/elasticsearch/stac_fastapi/elasticsearch/database_logic.py

@@ -1776,7 +1776,7 @@ async def get_all_catalogs(
             A tuple of (catalogs, next pagination token if any, optional count).
         """
         # Define sortable fields for catalogs
-        sortable_fields = ["id", "title"]
+        sortable_fields = ["id"]
 
         # Format the sort parameter
         formatted_sort = []

stac_fastapi/opensearch/stac_fastapi/opensearch/database_logic.py

@@ -1745,7 +1745,7 @@ async def get_all_catalogs(
             A tuple of (catalogs, next pagination token if any, optional count).
         """
         # Define sortable fields for catalogs
-        sortable_fields = ["id", "title"]
+        sortable_fields = ["id"]
 
         # Format the sort parameter
         formatted_sort = []

stac_fastapi/tests/extensions/test_catalogs.py

@@ -5,18 +5,17 @@
 
 @pytest.mark.asyncio
 async def test_get_root_catalog(catalogs_app_client, load_test_data):
-    """Test getting the root catalog."""
+    """Test getting the catalogs list."""
     resp = await catalogs_app_client.get("/catalogs")
     assert resp.status_code == 200
 
-    catalog = resp.json()
-    assert catalog["type"] == "Catalog"
-    assert catalog["id"] == "root"
-    assert catalog["stac_version"] == "1.0.0"
-    assert "links" in catalog
+    catalogs_response = resp.json()
+    assert "catalogs" in catalogs_response
+    assert "links" in catalogs_response
+    assert "numberReturned" in catalogs_response
 
-    # Check for required links
-    links = catalog["links"]
+    # Check for required pagination links
+    links = catalogs_response["links"]
     link_rels = [link["rel"] for link in links]
     assert "self" in link_rels
     assert "root" in link_rels
@@ -102,7 +101,7 @@ async def test_get_catalog_collections_nonexistent_catalog(catalogs_app_client):
 
 @pytest.mark.asyncio
 async def test_root_catalog_with_multiple_catalogs(catalogs_app_client, load_test_data):
-    """Test that root catalog includes links to multiple catalogs."""
+    """Test that catalogs response includes multiple catalog objects."""
     # Create multiple catalogs
     catalog_ids = []
     for i in range(3):
@@ -114,17 +113,17 @@ async def test_root_catalog_with_multiple_catalogs(catalogs_app_client, load_tes
         assert resp.status_code == 201
         catalog_ids.append(test_catalog["id"])
 
-    # Get root catalog
+    # Get catalogs list
     resp = await catalogs_app_client.get("/catalogs")
     assert resp.status_code == 200
 
-    catalog = resp.json()
-    child_links = [link for link in catalog["links"] if link["rel"] == "child"]
+    catalogs_response = resp.json()
+    returned_catalogs = catalogs_response["catalogs"]
+    returned_ids = [catalog["id"] for catalog in returned_catalogs]
 
-    # Should have child links for all created catalogs
-    child_hrefs = [link["href"] for link in child_links]
+    # Should have all created catalogs in the response
     for catalog_id in catalog_ids:
-        assert any(catalog_id in href for href in child_hrefs)
+        assert catalog_id in returned_ids
 
 
 @pytest.mark.asyncio
@@ -321,11 +320,12 @@ async def test_catalogs_pagination_limit(catalogs_app_client, load_test_data):
     resp = await catalogs_app_client.get("/catalogs?limit=2")
     assert resp.status_code == 200
 
-    catalog = resp.json()
-    child_links = [link for link in catalog["links"] if link["rel"] == "child"]
+    catalogs_response = resp.json()
+    returned_catalogs = catalogs_response["catalogs"]
 
-    # Should only return 2 child links
-    assert len(child_links) == 2
+    # Should only return 2 catalogs
+    assert len(returned_catalogs) == 2
+    assert catalogs_response["numberReturned"] == 2
 
 
 @pytest.mark.asyncio
@@ -346,11 +346,12 @@ async def test_catalogs_pagination_default_limit(catalogs_app_client, load_test_
     resp = await catalogs_app_client.get("/catalogs")
     assert resp.status_code == 200
 
-    catalog = resp.json()
-    child_links = [link for link in catalog["links"] if link["rel"] == "child"]
+    catalogs_response = resp.json()
+    returned_catalogs = catalogs_response["catalogs"]
 
-    # Should return default limit of 10 child links
-    assert len(child_links) == 10
+    # Should return default limit of 10 catalogs
+    assert len(returned_catalogs) == 10
+    assert catalogs_response["numberReturned"] == 10
 
 
 @pytest.mark.asyncio
@@ -375,6 +376,6 @@ async def test_catalogs_pagination_token_parameter(catalogs_app_client, load_tes
     resp = await catalogs_app_client.get("/catalogs?token=invalid_token")
     assert resp.status_code == 200
 
-    catalog = resp.json()
-    assert catalog["type"] == "Catalog"
-    assert "links" in catalog
+    catalogs_response = resp.json()
+    assert "catalogs" in catalogs_response
+    assert "links" in catalogs_response