route scratch

Author: jonhealy1

stac_fastapi/core/stac_fastapi/core/base_database_logic.py

@@ -138,3 +138,39 @@ async def delete_collection(
     ) -> None:
         """Delete a collection from the database."""
         pass
+
+    @abc.abstractmethod
+    async def get_all_catalogs(
+        self,
+        token: Optional[str],
+        limit: int,
+        request: Any = None,
+        sort: Optional[List[Dict[str, Any]]] = None,
+    ) -> Tuple[List[Dict[str, Any]], Optional[str], Optional[int]]:
+        """Retrieve a list of catalogs from the database, supporting pagination.
+
+        Args:
+            token (Optional[str]): The pagination token.
+            limit (int): The number of results to return.
+            request (Any, optional): The FastAPI request object. Defaults to None.
+            sort (Optional[List[Dict[str, Any]]], optional): Optional sort parameter. Defaults to None.
+
+        Returns:
+            A tuple of (catalogs, next pagination token if any, optional count).
+        """
+        pass
+
+    @abc.abstractmethod
+    async def create_catalog(self, catalog: Dict, refresh: bool = False) -> None:
+        """Create a catalog in the database."""
+        pass
+
+    @abc.abstractmethod
+    async def find_catalog(self, catalog_id: str) -> Dict:
+        """Find a catalog in the database."""
+        pass
+
+    @abc.abstractmethod
+    async def delete_catalog(self, catalog_id: str, refresh: bool = False) -> None:
+        """Delete a catalog from the database."""
+        pass

stac_fastapi/core/stac_fastapi/core/core.py

@@ -23,9 +23,14 @@
 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 CollectionSerializer, ItemSerializer
+from stac_fastapi.core.serializers import (
+    CatalogSerializer,
+    CollectionSerializer,
+    ItemSerializer,
+)
 from stac_fastapi.core.session import Session
 from stac_fastapi.core.utilities import filter_fields, get_bool_env
 from stac_fastapi.extensions.core.transaction import AsyncBaseTransactionsClient
@@ -82,6 +87,7 @@ class CoreClient(AsyncBaseCoreClient):
     collection_serializer: Type[CollectionSerializer] = attr.ib(
         default=CollectionSerializer
     )
+    catalog_serializer: Type[CatalogSerializer] = attr.ib(default=CatalogSerializer)
     post_request_model = attr.ib(default=BaseSearchPostRequest)
     stac_version: str = attr.ib(default=STAC_VERSION)
     landing_page_id: str = attr.ib(default="stac-fastapi")
@@ -142,15 +148,24 @@ def _landing_page(
         )
         return landing_page
 
-    async def landing_page(self, **kwargs) -> stac_types.LandingPage:
+    async def landing_page(self, **kwargs) -> Union[stac_types.LandingPage, Catalog]:
         """Landing page.
 
         Called with `GET /`.
 
         Returns:
-            API landing page, serving as an entry point to the API.
+            API landing page, serving as an entry point to the API, or root catalog if CatalogsExtension is enabled.
         """
         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,

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

@@ -1,22 +1,24 @@
 """Catalogs extension."""
 
-from typing import List, Type, Union
+from typing import List, Type
 
 import attr
-from fastapi import APIRouter, FastAPI, Request
+from fastapi import APIRouter, FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse
 from starlette.responses import Response
 
+from stac_fastapi.core.models import Catalog
+from stac_fastapi.types import stac as stac_types
 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.
+    The Catalogs extension adds a /catalogs endpoint that returns the root catalog
+    containing child links to all catalogs in the database.
     """
 
     client: BaseCoreClient = attr.ib(default=None)
@@ -25,40 +27,229 @@ class CatalogsExtension(ApiExtension):
     router: APIRouter = attr.ib(default=attr.Factory(APIRouter))
     response_class: Type[Response] = attr.ib(default=JSONResponse)
 
-    def register(self, app: FastAPI) -> None:
+    def register(self, app: FastAPI, settings=None) -> None:
         """Register the extension with a FastAPI application.
 
         Args:
             app: target FastAPI application.
-
-        Returns:
-            None
+            settings: extension settings (unused for now).
         """
-        response_model = (
-            self.settings.get("response_model")
-            if isinstance(self.settings, dict)
-            else getattr(self.settings, "response_model", None)
-        )
+        self.settings = settings or {}
 
         self.router.add_api_route(
             path="/catalogs",
             endpoint=self.catalogs,
             methods=["GET"],
-            response_model=LandingPage if response_model else None,
+            response_model=Catalog,
+            response_class=self.response_class,
+            summary="Get Root Catalog",
+            description="Returns the root catalog containing links to all catalogs.",
+            tags=["Catalogs"],
+        )
+
+        # Add endpoint for creating catalogs
+        self.router.add_api_route(
+            path="/catalogs",
+            endpoint=self.create_catalog,
+            methods=["POST"],
+            response_model=Catalog,
+            response_class=self.response_class,
+            status_code=201,
+            summary="Create Catalog",
+            description="Create a new STAC catalog.",
+            tags=["Catalogs"],
+        )
+
+        # Add endpoint for getting individual catalogs
+        self.router.add_api_route(
+            path="/catalogs/{catalog_id}",
+            endpoint=self.get_catalog,
+            methods=["GET"],
+            response_model=Catalog,
             response_class=self.response_class,
-            summary="Get Catalogs",
-            description="Returns the root catalog.",
+            summary="Get Catalog",
+            description="Get a specific STAC catalog by ID.",
             tags=["Catalogs"],
         )
+
+        # Add endpoint for getting collections in a catalog
+        self.router.add_api_route(
+            path="/catalogs/{catalog_id}/collections",
+            endpoint=self.get_catalog_collections,
+            methods=["GET"],
+            response_model=stac_types.Collections,
+            response_class=self.response_class,
+            summary="Get Catalog Collections",
+            description="Get collections linked from a specific catalog.",
+            tags=["Catalogs"],
+        )
+
         app.include_router(self.router, tags=["Catalogs"])
 
-    async def catalogs(self, request: Request) -> Union[LandingPage, Response]:
-        """Get catalogs.
+    async def catalogs(self, request: Request) -> Catalog:
+        """Get root catalog with links to all catalogs.
 
         Args:
             request: Request object.
 
         Returns:
-            The root catalog (landing page).
+            Root catalog containing child links to all catalogs in the database.
         """
-        return await self.client.landing_page(request=request)
+        base_url = str(request.base_url)
+
+        # Get all catalogs from database
+        catalogs, _, _ = await self.client.database.get_all_catalogs(
+            token=None,
+            limit=1000,  # Large limit to get all catalogs
+            request=request,
+            sort=[{"field": "id", "direction": "asc"}],
+        )
+
+        # Create child links to each catalog
+        child_links = []
+        for catalog in catalogs:
+            child_links.append(
+                {
+                    "rel": "child",
+                    "href": f"{base_url}catalogs/{catalog.id}",
+                    "type": "application/json",
+                    "title": catalog.title or catalog.id,
+                }
+            )
+
+        # 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)
+
+    async def create_catalog(self, catalog: Catalog, request: Request) -> Catalog:
+        """Create a new catalog.
+
+        Args:
+            catalog: The catalog to create.
+            request: Request object.
+
+        Returns:
+            The created catalog.
+        """
+        # Convert STAC catalog to database format
+        db_catalog = self.client.catalog_serializer.stac_to_db(catalog, request)
+
+        # Create the catalog in the database
+        await self.client.database.create_catalog(db_catalog.model_dump())
+
+        # Return the created catalog
+        return catalog
+
+    async def get_catalog(self, catalog_id: str, request: Request) -> Catalog:
+        """Get a specific catalog by ID.
+
+        Args:
+            catalog_id: The ID of the catalog to retrieve.
+            request: Request object.
+
+        Returns:
+            The requested catalog.
+        """
+        try:
+            # Get the catalog from the database
+            db_catalog = await self.client.database.find_catalog(catalog_id)
+
+            # Convert to STAC format
+            catalog = self.client.catalog_serializer.db_to_stac(db_catalog, request)
+
+            return catalog
+        except Exception:
+            raise HTTPException(
+                status_code=404, detail=f"Catalog {catalog_id} not found"
+            )
+
+    async def get_catalog_collections(
+        self, catalog_id: str, request: Request
+    ) -> stac_types.Collections:
+        """Get collections linked from a specific catalog.
+
+        Args:
+            catalog_id: The ID of the catalog.
+            request: Request object.
+
+        Returns:
+            Collections object containing collections linked from the catalog.
+        """
+        try:
+            # Get the catalog from the database
+            db_catalog = await self.client.database.find_catalog(catalog_id)
+
+            # Convert to STAC format to access links
+            catalog = self.client.catalog_serializer.db_to_stac(db_catalog, request)
+
+            # Extract collection IDs from catalog links
+            collection_ids = []
+            if hasattr(catalog, "links") and catalog.links:
+                for link in catalog.links:
+                    if link.get("rel") in ["child", "item"]:
+                        # Extract collection ID from href
+                        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)
+
+            # Fetch the collections
+            collections = []
+            for coll_id in collection_ids:
+                try:
+                    collection = await self.client.get_collection(
+                        coll_id, request=request
+                    )
+                    collections.append(collection)
+                except Exception:
+                    # Skip collections that can't be found
+                    continue
+
+            # Return in Collections format
+            base_url = str(request.base_url)
+            return stac_types.Collections(
+                collections=collections,
+                links=[
+                    {"rel": "root", "type": "application/json", "href": base_url},
+                    {"rel": "parent", "type": "application/json", "href": base_url},
+                    {
+                        "rel": "self",
+                        "type": "application/json",
+                        "href": f"{base_url}catalogs/{catalog_id}/collections",
+                    },
+                ],
+            )
+
+        except Exception:
+            raise HTTPException(
+                status_code=404, detail=f"Catalog {catalog_id} not found"
+            )

stac_fastapi/core/stac_fastapi/core/models/__init__.py

@@ -1 +1,27 @@
-"""stac_fastapi.elasticsearch.models module."""
+"""STAC models."""
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel
+
+
+class Catalog(BaseModel):
+    """STAC Catalog model."""
+
+    type: str = "Catalog"
+    stac_version: str
+    id: str
+    title: Optional[str] = None
+    description: Optional[str] = None
+    links: List[Dict[str, Any]]
+    stac_extensions: Optional[List[str]] = None
+
+
+class PartialCatalog(BaseModel):
+    """Partial STAC Catalog model for updates."""
+
+    id: str
+    title: Optional[str] = None
+    description: Optional[str] = None
+    links: Optional[List[Dict[str, Any]]] = None
+    stac_extensions: Optional[List[str]] = None