pagination, limit fix for catalogs

Author: jonhealy1

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

@@ -1,9 +1,10 @@
 """Catalogs extension."""
 
 from typing import List, Optional, Type
+from urllib.parse import urlparse
 
 import attr
-from fastapi import APIRouter, FastAPI, HTTPException, Request
+from fastapi import APIRouter, FastAPI, HTTPException, Query, Request
 from fastapi.responses import JSONResponse
 from starlette.responses import Response
 
@@ -122,21 +123,37 @@ def register(self, app: FastAPI, settings=None) -> None:
 
         app.include_router(self.router, tags=["Catalogs"])
 
-    async def catalogs(self, request: Request) -> Catalog:
+    async def catalogs(
+        self,
+        request: Request,
+        limit: Optional[int] = Query(
+            10,
+            ge=1,
+            description=(
+                "The maximum number of catalogs to return (page size). Defaults to 10."
+            ),
+        ),
+        token: Optional[str] = Query(
+            None,
+            description="Pagination token for the next page of results",
+        ),
+    ) -> Catalog:
         """Get root catalog with links to all catalogs.
 
         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.
         """
         base_url = str(request.base_url)
 
-        # Get all catalogs from database
+        # Get all catalogs from database with pagination
         catalogs, _, _ = await self.client.database.get_all_catalogs(
-            token=None,
-            limit=1000,  # Large limit to get all catalogs
+            token=token,
+            limit=limit,
             request=request,
             sort=[{"field": "id", "direction": "asc"}],
         )
@@ -258,15 +275,26 @@ async def get_catalog_collections(
             if hasattr(catalog, "links") and catalog.links:
                 for link in catalog.links:
                     if link.get("rel") in ["child", "item"]:
-                        # Extract collection ID from href
+                        # Extract collection ID from href using proper URL parsing
                         href = link.get("href", "")
-                        # Look for patterns like /collections/{id} or collections/{id}
-                        if "/collections/" in href:
-                            collection_id = href.split("/collections/")[-1].split("/")[
-                                0
-                            ]
-                            if collection_id and collection_id not in collection_ids:
-                                collection_ids.append(collection_id)
+                        if href:
+                            try:
+                                parsed_url = urlparse(href)
+                                path = parsed_url.path
+                                # Look for patterns like /collections/{id} or collections/{id}
+                                if "/collections/" in path:
+                                    # Split by /collections/ and take the last segment
+                                    path_parts = path.split("/collections/")
+                                    if len(path_parts) > 1:
+                                        collection_id = path_parts[1].split("/")[0]
+                                        if (
+                                            collection_id
+                                            and collection_id not in collection_ids
+                                        ):
+                                            collection_ids.append(collection_id)
+                            except Exception:
+                                # If URL parsing fails, skip this link
+                                continue
 
             # Fetch the collections
             collections = []

stac_fastapi/tests/extensions/test_catalogs.py

@@ -301,3 +301,80 @@ async def test_get_catalog_collection_item_nonexistent_item(
         f"/catalogs/{test_catalog['id']}/collections/{ctx.collection['id']}/items/nonexistent-item"
     )
     assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_catalogs_pagination_limit(catalogs_app_client, load_test_data):
+    """Test that pagination limit parameter works for catalogs endpoint."""
+    # Create multiple catalogs
+    catalog_ids = []
+    for i in range(5):
+        test_catalog = load_test_data("test_catalog.json")
+        test_catalog["id"] = f"test-catalog-{uuid.uuid4()}-{i}"
+        test_catalog["title"] = f"Test Catalog {i}"
+
+        resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+        assert resp.status_code == 201
+        catalog_ids.append(test_catalog["id"])
+
+    # Test with limit=2
+    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"]
+
+    # Should only return 2 child links
+    assert len(child_links) == 2
+
+
+@pytest.mark.asyncio
+async def test_catalogs_pagination_default_limit(catalogs_app_client, load_test_data):
+    """Test that pagination uses default limit when no limit parameter is provided."""
+    # Create multiple catalogs
+    catalog_ids = []
+    for i in range(15):
+        test_catalog = load_test_data("test_catalog.json")
+        test_catalog["id"] = f"test-catalog-{uuid.uuid4()}-{i}"
+        test_catalog["title"] = f"Test Catalog {i}"
+
+        resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+        assert resp.status_code == 201
+        catalog_ids.append(test_catalog["id"])
+
+    # Test without limit parameter (should default to 10)
+    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"]
+
+    # Should return default limit of 10 child links
+    assert len(child_links) == 10
+
+
+@pytest.mark.asyncio
+async def test_catalogs_pagination_limit_validation(catalogs_app_client):
+    """Test that pagination limit parameter validation works."""
+    # Test with limit=0 (should be invalid)
+    resp = await catalogs_app_client.get("/catalogs?limit=0")
+    assert resp.status_code == 400  # Validation error returns 400 for Query parameters
+
+
+@pytest.mark.asyncio
+async def test_catalogs_pagination_token_parameter(catalogs_app_client, load_test_data):
+    """Test that pagination token parameter is accepted (even if token is invalid)."""
+    # Create a catalog first
+    test_catalog = load_test_data("test_catalog.json")
+    test_catalog["id"] = f"test-catalog-{uuid.uuid4()}"
+
+    resp = await catalogs_app_client.post("/catalogs", json=test_catalog)
+    assert resp.status_code == 201
+
+    # Test with token parameter (even if invalid, should be accepted)
+    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