create MosaicJSON extension for / and /validate (#1257)

Author: vincentsarago

CHANGES.md

@@ -20,6 +20,10 @@
 * add `FsReader` which use `fs_open_dataset` as `dataset_opener`
 * create offical application `titiler.xarray.main:app`
 
+### titiler.mosaic
+
+* move `/` and `/validate`  to a `MosaicJSONExtension`
+
 ## 0.24.2 (2025-10-16)
 
 ### titiler.core

pyproject.toml

@@ -71,6 +71,9 @@ docs = [
     "griffe-inherited-docstrings>=1.0.0",
     "mkdocstrings[python]>=0.25.1",
 ]
+server = [
+    "uvicorn",
+]
 
 [project.urls]
 Homepage = 'https://developmentseed.org/titiler/'

src/titiler/application/titiler/application/main.py

@@ -42,6 +42,7 @@
     stacViewerExtension,
 )
 from titiler.mosaic.errors import MOSAIC_STATUS_CODES
+from titiler.mosaic.extensions import MosaicJSONExtension
 from titiler.mosaic.factory import MosaicTilerFactory
 
 logging.getLogger("botocore.credentials").disabled = True
@@ -170,6 +171,9 @@ def validate_access_token(access_token: str = Security(api_key_query)):
 if not api_settings.disable_mosaic:
     mosaic = MosaicTilerFactory(
         router_prefix="/mosaicjson",
+        extensions=[
+            MosaicJSONExtension(),
+        ],
         enable_telemetry=api_settings.telemetry_enabled,
         templates=titiler_templates,
     )

src/titiler/mosaic/tests/test_factory.py

@@ -20,6 +20,7 @@
 
 from titiler.core.dependencies import DefaultDependency
 from titiler.core.resources.enums import OptionalHeader
+from titiler.mosaic.extensions import MosaicJSONExtension
 from titiler.mosaic.factory import MosaicTilerFactory
 
 from .conftest import DATA_DIR
@@ -50,6 +51,15 @@ def test_MosaicTilerFactory():
         optional_headers=[OptionalHeader.x_assets],
         router_prefix="mosaic",
     )
+    assert len(mosaic.router.routes) == 15
+
+    mosaic = MosaicTilerFactory(
+        optional_headers=[OptionalHeader.x_assets],
+        extensions=[
+            MosaicJSONExtension(),
+        ],
+        router_prefix="mosaic",
+    )
     assert len(mosaic.router.routes) == 17
 
     app = FastAPI()

src/titiler/mosaic/titiler/mosaic/extensions.py

@@ -0,0 +1,59 @@
+"""titiler.mosaic extensions."""
+
+import logging
+from dataclasses import dataclass
+
+import rasterio
+from cogeo_mosaic.mosaic import MosaicJSON
+from fastapi import Depends
+
+from titiler.core.factory import FactoryExtension
+from titiler.mosaic.factory import MosaicTilerFactory
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class MosaicJSONExtension(FactoryExtension):
+    """Add MosaicJSON specific endpoints"""
+
+    def register(self, factory: MosaicTilerFactory):  # noqa: C901
+        """Register endpoint to the tiler factory."""
+
+        ############################################################################
+        # /read
+        ############################################################################
+
+        @factory.router.get(
+            "/",
+            response_model=MosaicJSON,
+            response_model_exclude_none=True,
+            responses={200: {"description": "Return MosaicJSON definition"}},
+            operation_id=f"{factory.operation_prefix}getMosaicJSON",
+        )
+        def read(
+            src_path=Depends(factory.path_dependency),
+            backend_params=Depends(factory.backend_dependency),
+            reader_params=Depends(factory.reader_dependency),
+            env=Depends(factory.environment_dependency),
+        ):
+            """Read a MosaicJSON"""
+            with rasterio.Env(**env):
+                logger.info(
+                    f"opening data with backend: {factory.backend} and reader {factory.dataset_reader}"
+                )
+                with factory.backend(
+                    src_path,
+                    reader=factory.dataset_reader,
+                    reader_options=reader_params.as_dict(),
+                    **backend_params.as_dict(),
+                ) as src_dst:
+                    return src_dst.mosaic_def
+
+        @factory.router.post(
+            "/validate",
+            operation_id=f"{factory.operation_prefix}validate",
+        )
+        def validate(body: MosaicJSON):
+            """Validate a MosaicJSON"""
+            return True

src/titiler/mosaic/titiler/mosaic/factory.py

@@ -9,7 +9,6 @@
 from attrs import define, field
 from cogeo_mosaic.backends import BaseBackend, MosaicBackend
 from cogeo_mosaic.models import Info as mosaicInfo
-from cogeo_mosaic.mosaic import MosaicJSON
 from fastapi import Depends, HTTPException, Path, Query
 from geojson_pydantic.features import Feature
 from geojson_pydantic.geometries import Polygon
@@ -143,7 +142,6 @@ class MosaicTilerFactory(BaseFactory):
     def register_routes(self):
         """This Method register routes to the router."""
 
-        self.read()
         self.info()
         self.tilesets()
         self.tile()
@@ -152,41 +150,8 @@ def register_routes(self):
         self.tilejson()
         self.wmts()
         self.point()
-        self.validate()
         self.assets()
 
-    ############################################################################
-    # /read
-    ############################################################################
-    def read(self):
-        """Register / (Get) Read endpoint."""
-
-        @self.router.get(
-            "/",
-            response_model=MosaicJSON,
-            response_model_exclude_none=True,
-            responses={200: {"description": "Return MosaicJSON definition"}},
-            operation_id=f"{self.operation_prefix}getMosaicJSON",
-        )
-        def read(
-            src_path=Depends(self.path_dependency),
-            backend_params=Depends(self.backend_dependency),
-            reader_params=Depends(self.reader_dependency),
-            env=Depends(self.environment_dependency),
-        ):
-            """Read a MosaicJSON"""
-            with rasterio.Env(**env):
-                logger.info(
-                    f"opening data with backend: {self.backend} and reader {self.dataset_reader}"
-                )
-                with self.backend(
-                    src_path,
-                    reader=self.dataset_reader,
-                    reader_options=reader_params.as_dict(),
-                    **backend_params.as_dict(),
-                ) as src_dst:
-                    return src_dst.mosaic_def
-
     ############################################################################
     # /info
     ############################################################################
@@ -1021,17 +986,6 @@ def point(
                 ],
             }
 
-    def validate(self):
-        """Register /validate endpoint."""
-
-        @self.router.post(
-            "/validate",
-            operation_id=f"{self.operation_prefix}validate",
-        )
-        def validate(body: MosaicJSON):
-            """Validate a MosaicJSON"""
-            return True
-
     def assets(self):
         """Register /assets endpoint."""