Convenience method to allow customizing route dependencies

stac_fastapi/api/stac_fastapi/api/app.py

@@ -1,10 +1,11 @@
 """fastapi app creation."""
-from typing import Any, Callable, Dict, List, Optional, Type, Union
+from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
 
 import attr
 from brotli_asgi import BrotliMiddleware
 from fastapi import APIRouter, FastAPI
 from fastapi.openapi.utils import get_openapi
+from fastapi.params import Depends
 from pydantic import BaseModel
 from stac_pydantic import Collection, Item, ItemCollection
 from stac_pydantic.api import ConformanceClasses, LandingPage, Search
@@ -22,7 +23,12 @@
     SearchGetRequest,
     _create_request_model,
 )
-from stac_fastapi.api.routes import create_async_endpoint, create_sync_endpoint
+from stac_fastapi.api.routes import (
+    Scope,
+    add_route_dependencies,
+    create_async_endpoint,
+    create_sync_endpoint,
+)
 
 # TODO: make this module not depend on `stac_fastapi.extensions`
 from stac_fastapi.extensions.core import FieldsExtension
@@ -77,6 +83,7 @@ class StacApi:
     item_collection_uri: Type[ItemCollectionUri] = attr.ib(default=ItemCollectionUri)
     response_class: Type[Response] = attr.ib(default=JSONResponse)
     middlewares: List = attr.ib(default=attr.Factory(lambda: [BrotliMiddleware]))
+    route_dependencies: List[Tuple[List[Scope], List[Depends]]] = attr.ib(default=[])
 
     def get_extension(self, extension: Type[ApiExtension]) -> Optional[ApiExtension]:
         """Get an extension.
@@ -312,6 +319,12 @@ async def ping():
 
         self.app.include_router(mgmt_router, tags=["Liveliness/Readiness"])
 
+    def add_route_dependencies(
+        self, scopes: List[Scope], dependencies=List[Depends]
+    ) -> None:
+        """Add custom dependencies to routes."""
+        return add_route_dependencies(self.router.routes, scopes, dependencies)
+
     def __attrs_post_init__(self):
         """Post-init hook.
 
@@ -353,3 +366,7 @@ def __attrs_post_init__(self):
         # add middlewares
         for middleware in self.middlewares:
             self.app.add_middleware(middleware)
+
+        # customize route dependencies
+        for scopes, dependencies in self.route_dependencies:
+            self.add_route_dependencies(scopes=scopes, dependencies=dependencies)

stac_fastapi/api/stac_fastapi/api/routes.py

@@ -1,10 +1,12 @@
 """route factories."""
-from typing import Any, Callable, Dict, Type, Union
+from typing import Any, Callable, Dict, List, Optional, Type, TypedDict, Union
 
-from fastapi import Depends
+from fastapi import APIRouter, Depends
+from fastapi.dependencies.utils import get_parameterless_sub_dependant
 from pydantic import BaseModel
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
+from starlette.routing import BaseRoute, Match
 
 from stac_fastapi.api.models import APIRequest
 
@@ -94,3 +96,40 @@ def _endpoint(
             return _wrap_response(func(request_data, request=request), response_class)
 
     return _endpoint
+
+
+class Scope(TypedDict, total=False):
+    # More strict version of Starlette's Scope
+    # https://github.com/encode/starlette/blob/6af5c515e0a896cbf3f86ee043b88f6c24200bcf/starlette/types.py#L3
+    path: str
+    method: str
+    type: Optional[str]
+
+
+def add_route_dependencies(
+    routes: List[BaseRoute], scopes: List[Scope], dependencies=List[Depends]
+) -> None:
+    """Add dependencies to routes.
+
+    Allows a developer to add dependencies to a route after the route has been
+    defined.
+
+    Returns:
+        None
+    """
+    for scope in scopes:
+        for route in routes:
+
+            match, _ = route.matches({"type": "http", **scope})
+            if match != Match.FULL:
+                continue
+
+            # Mimicking how APIRoute handles dependencies:
+            # https://github.com/tiangolo/fastapi/blob/1760da0efa55585c19835d81afa8ca386036c325/fastapi/routing.py#L408-L412
+            for depends in dependencies[::-1]:
+                route.dependant.dependencies.insert(
+                    0,
+                    get_parameterless_sub_dependant(
+                        depends=depends, path=route.path_format
+                    ),
+                )