feat: add redis cache for navigation

Author: Yuri Zmytrakov

.github/workflows/cicd.yml

@@ -66,6 +66,16 @@ jobs:
         ports:
           - 9202:9202
 
+      redis:
+        image: redis:7-alpine
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 6379:6379
+
     strategy:
       matrix:
         python-version: [ "3.9", "3.10", "3.11", "3.12", "3.13"]
@@ -126,3 +136,6 @@ jobs:
           DATABASE_REFRESH: true
           ES_VERIFY_CERTS: false
           BACKEND: ${{ matrix.backend == 'elasticsearch7' && 'elasticsearch' || matrix.backend == 'elasticsearch8' && 'elasticsearch' || 'opensearch' }}
+          REDIS_ENABLE: true
+          REDIS_HOST: localhost
+          REDIS_PORT: 6379

Makefile

@@ -82,7 +82,7 @@ test-datetime-filtering-os:
 	docker compose down
 
 .PHONY: test
-test: test-elasticsearch test-datetime-filtering-es test-opensearch test-datetime-filtering-os
+test: test-elasticsearch test-datetime-filtering-es test-opensearch test-datetime-filtering-os test-redis-es test-redis-os
 
 .PHONY: run-database-es
 run-database-es:
@@ -117,4 +117,16 @@ docs-image:
 .PHONY: docs
 docs: docs-image
 	docker compose -f compose.docs.yml \
-		run docs
+		run docs
+
+.PHONY: test-redis-es
+test-redis-es:
+	docker compose -f compose-redis.yml up -d
+	-$(run_es) /bin/bash -c 'export REDIS_ENABLE=true REDIS_HOST=redis REDIS_PORT=6379 && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest redis/ -v'
+	docker compose -f compose-redis.yml down
+
+.PHONY: test-redis-os
+test-redis-os:
+	docker compose -f compose-redis.yml up -d
+	-$(run_os) /bin/bash -c 'export REDIS_ENABLE=true REDIS_HOST=redis REDIS_PORT=6379 && ./scripts/wait-for-it-es.sh opensearch:9202 && cd stac_fastapi/tests/ && pytest redis/ -v'
+	docker compose -f compose-redis.yml down

compose-redis.yml

@@ -0,0 +1,13 @@
+version: '3.8'
+
+services:
+  redis:
+    image: redis:7-alpine
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_test_data:/data
+    command: redis-server --appendonly yes
+
+volumes:
+  redis_test_data:

stac_fastapi/core/stac_fastapi/core/core.py

@@ -24,9 +24,10 @@
 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, 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
+from stac_fastapi.core.utilities import filter_fields, get_bool_env
 from stac_fastapi.extensions.core.transaction import AsyncBaseTransactionsClient
 from stac_fastapi.extensions.core.transaction.request import (
     PartialCollection,
@@ -328,6 +329,15 @@ async def all_collections(
             if parsed_sort:
                 sort = parsed_sort
 
+        current_url = str(request.url)
+        redis_enable = get_bool_env("REDIS_ENABLE", default=False)
+
+        if redis_enable:
+            try:
+                redis = await connect_redis()
+            except Exception:
+                redis = None
+
         # Convert q to a list if it's a string
         q_list = None
         if q is not None:
@@ -426,6 +436,22 @@ async def all_collections(
             },
         ]
 
+        if redis_enable and 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": "prev",
+                        "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 +770,14 @@ async def post_search(
             HTTPException: If there is an error with the cql2_json filter.
         """
         base_url = str(request.base_url)
+        redis_enable = get_bool_env("REDIS_ENABLE", default=False)
+
+        redis = None
+        if redis_enable:
+            try:
+                redis = await connect_redis()
+            except Exception:
+                redis = None
 
         search = self.database.make_search()
 
@@ -850,6 +884,41 @@ async def post_search(
         ]
         links = await PagingLinks(request=request, next=next_token).get_links()
 
+        collection_links = []
+        if search_request.collections:
+            for collection_id in search_request.collections:
+                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_enable and 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:
+                links.insert(
+                    0,
+                    {
+                        "rel": "prev",
+                        "type": "application/json",
+                        "method": "GET",
+                        "href": prev_link,
+                    },
+                )
+
         return stac_types.ItemCollection(
             type="FeatureCollection",
             features=items,

stac_fastapi/core/stac_fastapi/core/redis_utils.py

@@ -0,0 +1,124 @@
+"""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 = 15
+
+    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 Sentinel."""
+
+    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 = RedisSettings()
+
+
+async def connect_redis(settings: Optional[RedisSettings] = None) -> aioredis.Redis:
+    """Return a 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 connect_redis_sentinel(
+    settings: Optional[RedisSentinelSettings] = None,
+) -> Optional[aioredis.Redis]:
+    """Return a 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(
+                [(h, p) for h, p 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 save_self_link(
+    redis: aioredis.Redis, token: Optional[str], self_href: str
+) -> None:
+    """Save the self link for the current token with 30 min TTL."""
+    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]:
+    """Get the previous page link for the current token (if exists)."""
+    if not token:
+        return None
+    return await redis.get(f"nav:self:{token}")