add redis

Author: Yuri Zmytrakov

Makefile

@@ -63,7 +63,7 @@ docker-shell-os:
 
 .PHONY: test-elasticsearch
 test-elasticsearch:
-	-$(run_es) /bin/bash -c 'export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest'
+	-$(run_es) /bin/bash -c 'pip install redis export && ./scripts/wait-for-it-es.sh elasticsearch:9200 && cd stac_fastapi/tests/ && pytest'
 	docker compose down
 
 .PHONY: test-opensearch

compose.yml

@@ -22,6 +22,8 @@ services:
       - ES_VERIFY_CERTS=false
       - BACKEND=elasticsearch
       - DATABASE_REFRESH=true
+      - REDIS_HOST=redis
+      - REDIS_PORT=6379
     ports:
       - "8080:8080"
     volumes:
@@ -30,6 +32,7 @@ services:
       - ./esdata:/usr/share/elasticsearch/data
     depends_on:
       - elasticsearch
+      - redis
     command:
       bash -c "./scripts/wait-for-it-es.sh es-container:9200 && python -m stac_fastapi.elasticsearch.app"
 
@@ -94,3 +97,10 @@ services:
       - ./opensearch/snapshots:/usr/share/opensearch/snapshots
     ports:
       - "9202:9202"
+
+  redis:
+    container_name: stac-redis
+    image: redis:7.2-alpine
+    restart: always
+    ports:
+      - "6379:6379"

docker-compose.redis.yml

@@ -0,0 +1,24 @@
+version: '3.8'
+
+services:
+  redis:
+    image: redis:7-alpine
+    container_name: stac-app-redis
+    ports:
+      - "6379:6379"
+    volumes:
+      - redis_data:/data
+    command: >
+      redis-server
+      --appendonly yes
+      --maxmemory 256mb
+      --maxmemory-policy allkeys-lru
+    healthcheck:
+      test: ["CMD", "redis-cli", "ping"]
+      interval: 10s
+      timeout: 5s
+      retries: 3
+    restart: unless-stopped
+
+volumes:
+  redis_data:

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

@@ -44,6 +44,7 @@
 from stac_fastapi.types.extension import ApiExtension
 from stac_fastapi.types.requests import get_base_url
 from stac_fastapi.types.search import BaseSearchPostRequest
+from stac_fastapi.core.redis_utils import connect_redis, close_redis
 
 logger = logging.getLogger(__name__)
 
@@ -237,6 +238,18 @@ async def all_collections(self, **kwargs) -> stac_types.Collections:
         base_url = str(request.base_url)
         limit = int(request.query_params.get("limit", os.getenv("STAC_ITEM_LIMIT", 10)))
         token = request.query_params.get("token")
+        current_url = str(request.url)
+        redis = None
+        try:
+            redis = await connect_redis()
+            
+            if redis:
+                current_key = "current:collections"
+                await redis.setex(current_key, 600, current_url)
+                
+        except Exception as e:
+            logger.error(f"Redis connection error: {e}")
+            redis = None
 
         collections, next_token = await self.database.get_all_collections(
             token=token, limit=limit, request=request
@@ -252,10 +265,27 @@ async def all_collections(self, **kwargs) -> stac_types.Collections:
             },
         ]
 
-        if next_token:
+        if token and redis:
+            prev_key = "prev:collections"
+            previous_url = await redis.get(prev_key)
+
+            if previous_url:
+                if previous_url != current_url:
+                    links.append({
+                        "rel": "previous",
+                        "type": MimeTypes.json,
+                        "href": previous_url,
+                    })
+
+        if next_token and redis:
+            prev_key = "prev:collections"
+            await redis.setex(prev_key, 600, current_url)
+
             next_link = PagingLinks(next=next_token, request=request).link_next()
             links.append(next_link)
 
+        if redis:
+            await redis.close()
         return stac_types.Collections(collections=collections, links=links)
 
     async def get_collection(
@@ -310,20 +340,23 @@ async def item_collection(
         """
         request: Request = kwargs["request"]
         token = request.query_params.get("token")
-        if not hasattr(self, '_prev_links'):
-            self._prev_links = {}
-        
-        session_id = request.cookies.get('stac_session', 'default_session')
-        current_self_link = str(request.url)
-
-        if session_id not in self._prev_links:
-            self._prev_links[session_id] = []
-        
-        history = self._prev_links[session_id]
-        if not history or current_self_link != history[-1]:
-            history.append(current_self_link)
         base_url = str(request.base_url)
 
+        current_url = str(request.url)
+
+        try:
+            redis = await connect_redis()
+        except Exception as e:
+            logger.error(f"Redis connection error: {e}")
+            redis = None
+
+        if redis:
+            try:
+                current_key = f"current:{collection_id}"
+                await redis.setex(current_key, 600, current_url)
+            except Exception as e:
+                logger.error(f"Redis error: {e}")
+
         collection = await self.get_collection(
             collection_id=collection_id, request=request
         )
@@ -381,16 +414,28 @@ async def item_collection(
         ]
 
         paging_links = await PagingLinks(request=request, next=next_token).get_links()
-        history = self._prev_links.get(session_id, [])
-        if len(history) > 1:
-            previous_self_link = history[-2]
-            paging_links.append({
-                "rel": "previous",
-                "type": "application/json",
-                "href": previous_self_link,
-            })        
+        
+        if token and redis:
+            prev_key = f"prev:{collection_id}"
+            previous_url = await redis.get(prev_key)
+            
+            if previous_url:
+                # prevent looped navigation
+                if previous_url != current_url:
+                    paging_links.append({
+                        "rel": "previous",
+                        "type": "application/json",
+                        "href": previous_url,
+                    })
+        if redis and next_token:
+            prev_key = f"prev:{collection_id}"
+            await redis.setex(prev_key, 600, current_url)
+
         links = collection_links + paging_links
 
+        if redis:
+            await redis.close()
+
         return stac_types.ItemCollection(
             type="FeatureCollection",
             features=items,
@@ -529,6 +574,12 @@ async def post_search(
             HTTPException: If there is an error with the cql2_json filter.
         """
         base_url = str(request.base_url)
+        current_url = str(request.url)
+        try:
+            redis = await connect_redis()
+        except Exception as e:
+            logger.error(f"Redis connection error: {e}")
+            redis = None
 
         search = self.database.make_search()
 
@@ -628,6 +679,24 @@ async def post_search(
         ]
         links = await PagingLinks(request=request, next=next_token).get_links()
 
+        if search_request.token and redis:
+            prev_key = f"prev:search_result"
+            previous_url = await redis.get(prev_key)
+
+            if previous_url and previous_url != current_url:
+                links.append({
+                    "rel": "previous",
+                    "type": "application/json",
+                    "href": previous_url,
+                })
+
+        if redis and next_token:
+            prev_key = f"prev:search_result"
+            await redis.setex(prev_key, 600, current_url)
+
+        if redis:
+            await redis.close()
+
         return stac_types.ItemCollection(
             type="FeatureCollection",
             features=items,

stac_fastapi/core/stac_fastapi/core/redis_utils.py

@@ -0,0 +1,40 @@
+import os
+from redis import asyncio as aioredis
+from pydantic_settings import BaseSettings
+from stac_fastapi.core.utilities import get_bool_env
+from typing import Optional
+
+redis_pool = None
+
+class RedisSettings(BaseSettings):
+    url: str = os.getenv("REDIS_URL", "redis://redis:6379")
+    max_connections: int = int(os.getenv("REDIS_MAX_CONNECTIONS", 10))
+    retry_on_timeout: bool = get_bool_env("REDIS_RETRY_TIMEOUT", True)
+    decode_responses: bool = get_bool_env("REDIS_DECODE_RESPONSES", True)
+    client_name: str = os.getenv("REDIS_CLIENT_NAME", "stac-fastapi-app")
+    health_check_interval: int = int(os.getenv("REDIS_HEALTH_CHECK_INTERVAL", 30))
+
+redis_settings = RedisSettings()
+
+async def connect_redis(settings: Optional[RedisSettings] = None):
+    """Get Redis connection with optional custom settings"""
+    global redis_pool
+    settings = settings or redis_settings
+    
+    if redis_pool is None:
+        redis_pool = aioredis.ConnectionPool.from_url(
+            settings.url,
+            max_connections=settings.max_connections,
+            retry_on_timeout=settings.retry_on_timeout,
+            decode_responses=settings.decode_responses,
+            client_name=settings.client_name,
+            health_check_interval=settings.health_check_interval,
+        )
+    return aioredis.Redis(connection_pool=redis_pool)
+
+async def close_redis():
+    """Close Redis connection pool"""
+    global redis_pool
+    if redis_pool:
+        await redis_pool.disconnect()
+        redis_pool = None