feat: Add Redis caching for navigation

Implement Redis caching to support proper pagination navigation in STAC FastAPI.
- Adds Redis configuration for both Sentinel and standalone Redis setups
- Caches pagination tokens
- Enables prev/next links in paginated responses

Environment variables provided for flexible deployment configurations.

Author: Yuri Zmytrakov

.pre-commit-config.yaml

@@ -31,7 +31,8 @@ repos:
         ]
         additional_dependencies: [
           "types-attrs",
-          "types-requests"
+          "types-requests",
+          "types-redis"
         ]
   - repo: https://github.com/PyCQA/pydocstyle
     rev: 6.1.1

dockerfiles/Dockerfile.dev.es

@@ -18,3 +18,4 @@ COPY . /app
 RUN pip install --no-cache-dir -e ./stac_fastapi/core
 RUN pip install --no-cache-dir -e ./stac_fastapi/sfeos_helpers
 RUN pip install --no-cache-dir -e ./stac_fastapi/elasticsearch[dev,server]
+RUN pip install --no-cache-dir redis types-redis

stac_fastapi/core/setup.py

@@ -19,6 +19,7 @@
     "pygeofilter~=0.3.1",
     "jsonschema~=4.0.0",
     "slowapi~=0.1.9",
+    "redis==6.4.0",
 ]
 
 setup(

stac_fastapi/core/stac_fastapi/core/core.py

@@ -24,6 +24,11 @@
 from stac_fastapi.core.base_settings import ApiBaseSettings
 from stac_fastapi.core.datetime_utils import format_datetime_range
 from stac_fastapi.core.models.links import PagingLinks
+from stac_fastapi.core.redis_utils import (
+    connect_redis_sentinel,
+    get_prev_link,
+    save_self_link,
+)
 from stac_fastapi.core.serializers import CollectionSerializer, ItemSerializer
 from stac_fastapi.core.session import Session
 from stac_fastapi.core.utilities import filter_fields
@@ -268,6 +273,7 @@ async def all_collections(
         Returns:
             A Collections object containing all the collections in the database and links to various resources.
         """
+        request = kwargs["request"]
         base_url = str(request.base_url)
 
         # Get the global limit from environment variable
@@ -333,6 +339,12 @@ async def all_collections(
         if q is not None:
             q_list = [q] if isinstance(q, str) else q
 
+        current_url = str(request.url)
+        redis = None
+        try:
+            redis = await connect_redis_sentinel()
+        except Exception:
+            redis = None
         # Parse the query parameter if provided
         parsed_query = None
         if query is not None:
@@ -426,6 +438,21 @@ async def all_collections(
             },
         ]
 
+        if redis:
+            if next_token:
+                await save_self_link(redis, next_token, current_url)
+
+            prev_link = await get_prev_link(redis, token)
+            if prev_link:
+                links.insert(
+                    0,
+                    {
+                        "rel": "previous",
+                        "type": "application/json",
+                        "method": "GET",
+                        "href": prev_link,
+                    },
+                )
         if next_token:
             next_link = PagingLinks(next=next_token, request=request).link_next()
             links.append(next_link)
@@ -744,6 +771,10 @@ async def post_search(
             HTTPException: If there is an error with the cql2_json filter.
         """
         base_url = str(request.base_url)
+        try:
+            redis = await connect_redis_sentinel()
+        except Exception:
+            redis = None
 
         search = self.database.make_search()
 
@@ -850,6 +881,60 @@ async def post_search(
         ]
         links = await PagingLinks(request=request, next=next_token).get_links()
 
+        collection_links = []
+        if (
+            items
+            and search_request.collections
+            and len(search_request.collections) == 1
+        ):
+            collection_id = search_request.collections[0]
+            collection_links.extend(
+                [
+                    {
+                        "rel": "collection",
+                        "type": "application/json",
+                        "href": urljoin(base_url, f"collections/{collection_id}"),
+                    },
+                    {
+                        "rel": "parent",
+                        "type": "application/json",
+                        "href": urljoin(base_url, f"collections/{collection_id}"),
+                    },
+                ]
+            )
+        links.extend(collection_links)
+
+        if redis:
+            self_link = str(request.url)
+            await save_self_link(redis, next_token, self_link)
+
+            prev_link = await get_prev_link(redis, token_param)
+            if prev_link:
+                method = "GET"
+                body = None
+                for link in links:
+                    if link.get("rel") == "next":
+                        if "method" in link:
+                            method = link["method"]
+                        if "body" in link:
+                            body = {**link["body"]}
+                            body.pop("token", None)
+                        break
+                else:
+                    method = request.method
+
+                prev_link_data = {
+                    "rel": "previous",
+                    "type": "application/json",
+                    "method": method,
+                    "href": prev_link,
+                }
+
+                if body:
+                    prev_link_data["body"] = body
+
+                links.insert(0, prev_link_data)
+
         return stac_types.ItemCollection(
             type="FeatureCollection",
             features=items,

stac_fastapi/core/stac_fastapi/core/redis_utils.py

@@ -0,0 +1,123 @@
+"""Utilities for connecting to and managing Redis connections."""
+
+from typing import Optional
+
+from pydantic_settings import BaseSettings
+from redis import asyncio as aioredis
+from redis.asyncio.sentinel import Sentinel
+
+redis_pool: Optional[aioredis.Redis] = None
+
+
+class RedisSentinelSettings(BaseSettings):
+    """Configuration for connecting to Redis Sentinel."""
+
+    REDIS_SENTINEL_HOSTS: str = ""
+    REDIS_SENTINEL_PORTS: str = "26379"
+    REDIS_SENTINEL_MASTER_NAME: str = "master"
+    REDIS_DB: int = 0
+
+    REDIS_MAX_CONNECTIONS: int = 10
+    REDIS_RETRY_TIMEOUT: bool = True
+    REDIS_DECODE_RESPONSES: bool = True
+    REDIS_CLIENT_NAME: str = "stac-fastapi-app"
+    REDIS_HEALTH_CHECK_INTERVAL: int = 30
+
+
+class RedisSettings(BaseSettings):
+    """Configuration for connecting Redis."""
+
+    REDIS_HOST: str = ""
+    REDIS_PORT: int = 6379
+    REDIS_DB: int = 0
+
+    REDIS_MAX_CONNECTIONS: int = 10
+    REDIS_RETRY_TIMEOUT: bool = True
+    REDIS_DECODE_RESPONSES: bool = True
+    REDIS_CLIENT_NAME: str = "stac-fastapi-app"
+    REDIS_HEALTH_CHECK_INTERVAL: int = 30
+
+
+# Select the Redis or Redis Sentinel configuration
+redis_settings: BaseSettings = RedisSentinelSettings()
+
+
+async def connect_redis_sentinel(
+    settings: Optional[RedisSentinelSettings] = None,
+) -> Optional[aioredis.Redis]:
+    """Return Redis Sentinel connection."""
+    global redis_pool
+    settings = settings or redis_settings
+
+    if (
+        not settings.REDIS_SENTINEL_HOSTS
+        or not settings.REDIS_SENTINEL_PORTS
+        or not settings.REDIS_SENTINEL_MASTER_NAME
+    ):
+        return None
+
+    hosts = [h.strip() for h in settings.REDIS_SENTINEL_HOSTS.split(",") if h.strip()]
+    ports = [
+        int(p.strip()) for p in settings.REDIS_SENTINEL_PORTS.split(",") if p.strip()
+    ]
+
+    if redis_pool is None:
+        try:
+            sentinel = Sentinel(
+                [(host, port) for host, port in zip(hosts, ports)],
+                decode_responses=settings.REDIS_DECODE_RESPONSES,
+            )
+            master = sentinel.master_for(
+                service_name=settings.REDIS_SENTINEL_MASTER_NAME,
+                db=settings.REDIS_DB,
+                decode_responses=settings.REDIS_DECODE_RESPONSES,
+                retry_on_timeout=settings.REDIS_RETRY_TIMEOUT,
+                client_name=settings.REDIS_CLIENT_NAME,
+                max_connections=settings.REDIS_MAX_CONNECTIONS,
+                health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL,
+            )
+            redis_pool = master
+
+        except Exception:
+            return None
+
+    return redis_pool
+
+
+async def connect_redis(settings: Optional[RedisSettings] = None) -> aioredis.Redis:
+    """Return Redis connection."""
+    global redis_pool
+    settings = settings or redis_settings
+
+    if not settings.REDIS_HOST or not settings.REDIS_PORT:
+        return None
+
+    if redis_pool is None:
+        pool = aioredis.ConnectionPool(
+            host=settings.REDIS_HOST,
+            port=settings.REDIS_PORT,
+            db=settings.REDIS_DB,
+            max_connections=settings.REDIS_MAX_CONNECTIONS,
+            decode_responses=settings.REDIS_DECODE_RESPONSES,
+            retry_on_timeout=settings.REDIS_RETRY_TIMEOUT,
+            health_check_interval=settings.REDIS_HEALTH_CHECK_INTERVAL,
+        )
+        redis_pool = aioredis.Redis(
+            connection_pool=pool, client_name=settings.REDIS_CLIENT_NAME
+        )
+    return redis_pool
+
+
+async def save_self_link(
+    redis: aioredis.Redis, token: Optional[str], self_href: str
+) -> None:
+    """Add the self link for next page as prev link for the current token."""
+    if token:
+        await redis.setex(f"nav:self:{token}", 1800, self_href)
+
+
+async def get_prev_link(redis: aioredis.Redis, token: Optional[str]) -> Optional[str]:
+    """Pull the prev page link for the current token."""
+    if not token:
+        return None
+    return await redis.get(f"nav:self:{token}")