Implement Valkey caching layer and update related configurations

- Add CacheClient for managing Valkey interactions
- Integrate caching into bus API routes for improved performance
- Update .env.example and config.py for Valkey settings
- Modify Makefile and README for Valkey service management
- Add Valkey service to docker-compose
- Include Valkey dependency in pyproject.toml and uv.lock

.env.example

@@ -7,3 +7,7 @@ DATAMALL_BASE_URL=https://datamall2.mytransport.sg/ltaodataservice
 REQUEST_TIMEOUT_SECONDS=10
 MAX_CONNECTIONS=200
 MAX_KEEPALIVE_CONNECTIONS=100
+VALKEY_ENABLED=true
+VALKEY_URL=redis://valkey:6379/0
+VALKEY_CONNECT_TIMEOUT_SECONDS=1
+VALKEY_DEFAULT_TTL_SECONDS=120

AGENTS.md

@@ -46,6 +46,7 @@ The app fails to start if DATAMALL_API_KEY is missing.
 - App entrypoint and lifespan wiring: [app/main.py](app/main.py)
 - Settings and env loading: [app/core/config.py](app/core/config.py)
 - Upstream HTTP client wrapper: [app/services/lta_client.py](app/services/lta_client.py)
+- Valkey cache client wrapper: [app/services/cache.py](app/services/cache.py)
 - Bus API routes: [app/api/routes/bus.py](app/api/routes/bus.py)
 - Health routes: [app/api/routes/health.py](app/api/routes/health.py)
 - Container build: [Dockerfile](Dockerfile)
@@ -56,6 +57,7 @@ The app fails to start if DATAMALL_API_KEY is missing.
 - Keep endpoint namespace under /api/v1.
 - Preserve upstream query parameter names using aliases such as BusStopCode, ServiceNo, Date, and $skip.
 - Route handlers return upstream JSON payloads directly unless feature work requires transformation.
+- Keep cache reads/writes inside route helper flow; do not duplicate cache key logic across handlers.
 - Centralize outbound DataMall logic in [app/services/lta_client.py](app/services/lta_client.py), not inside routes.
 - Add new environment variables in [app/core/config.py](app/core/config.py) and document them in [.env.example](.env.example).
 

Makefile

@@ -10,7 +10,8 @@ COMPOSE_CMD:=$(shell if command -v podman >/dev/null 2>&1; then if podman compos
 .DEFAULT_GOAL := help
 
 .PHONY: help check-engine check-compose sync run dev prod compile clean \
-	image-build image-run image-stop compose-up compose-down compose-logs compose-ps
+	image-build image-run image-stop compose-up compose-down compose-logs compose-ps \
+	valkey-up valkey-down
 
 help:
 	@echo "Targets:"
@@ -27,6 +28,8 @@ help:
 	@echo "  compose-down Stop compose stack"
 	@echo "  compose-logs Follow compose logs"
 	@echo "  compose-ps   Show compose services"
+	@echo "  valkey-up    Start only Valkey service"
+	@echo "  valkey-down  Stop and remove only Valkey service"
 	@echo ""
 	@echo "Detected container engine: $(if $(CONTAINER_ENGINE),$(CONTAINER_ENGINE),none)"
 	@echo "Detected compose command: $(if $(COMPOSE_CMD),$(COMPOSE_CMD),none)"
@@ -84,3 +87,10 @@ compose-logs: check-compose
 
 compose-ps: check-compose
 	$(COMPOSE_CMD) ps
+
+valkey-up: check-compose
+	$(COMPOSE_CMD) up -d valkey
+
+valkey-down: check-compose
+	$(COMPOSE_CMD) stop valkey
+	$(COMPOSE_CMD) rm -f valkey

README.md

@@ -5,6 +5,7 @@ FastAPI backend platform for LTA DataMall Bus Transport APIs, dockerized for pro
 ## Features
 
 - Async FastAPI service with shared `httpx` connection pool
+- Valkey caching layer with graceful fallback when cache is unavailable
 - Scalable runtime with Gunicorn + Uvicorn workers
 - API key loaded from `.env` (`DATAMALL_API_KEY`)
 - Bus Transport endpoints exposed under `/api/v1`
@@ -51,6 +52,25 @@ docker compose up --build -d
 curl http://localhost:8000/healthz
 ```
 
+The compose stack includes a Valkey service for caching.
+
+## Cache Configuration
+
+Set in `.env`:
+
+- `VALKEY_ENABLED=true`
+- `VALKEY_URL=redis://valkey:6379/0`
+- `VALKEY_CONNECT_TIMEOUT_SECONDS=1`
+- `VALKEY_DEFAULT_TTL_SECONDS=120`
+
+Current cache behavior:
+
+- Bus Arrival: 15s
+- Bus Services and Bus Routes: 300s
+- Bus Stops: 1800s
+- Planned Bus Routes: 900s
+- Passenger Volume endpoints: 21600s
+
 ## Makefile Tasks
 
 Use `make help` to see all targets. The Makefile auto-detects Podman first, then Docker.
@@ -63,6 +83,8 @@ make run
 make compile
 make compose-up
 make compose-down
+make valkey-up
+make valkey-down
 ```
 
 ## Horizontal Scaling

app/api/routes/bus.py

@@ -3,6 +3,7 @@ from typing import Any
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
 
+from app.services.cache import CacheClient
 from app.services.lta_client import LTAClient
 
 router = APIRouter(prefix="/api/v1", tags=["Bus"])
@@ -12,6 +13,10 @@ async def get_lta_client(request: Request) -> LTAClient:
     return request.app.state.lta_client
 
 
+async def get_cache_client(request: Request) -> CacheClient:
+    return request.app.state.cache_client
+
+
 def _map_httpx_error(err: httpx.HTTPStatusError) -> HTTPException:
     return HTTPException(
         status_code=err.response.status_code,
@@ -23,17 +28,35 @@ def _map_httpx_error(err: httpx.HTTPStatusError) -> HTTPException:
     )
 
 
+async def _fetch_with_cache(
+    path: str,
+    params: dict[str, Any],
+    client: LTAClient,
+    cache: CacheClient,
+    ttl_seconds: int,
+) -> dict[str, Any]:
+    cache_key = cache.build_cache_key(path, params)
+    cached = await cache.get_json(cache_key)
+    if cached is not None:
+        return cached
+
+    payload = await client.get(path, params=params)
+    await cache.set_json(cache_key, payload, ttl_seconds=ttl_seconds)
+    return payload
+
+
 @router.get("/bus-arrival")
 async def bus_arrival(
     bus_stop_code: str = Query(..., alias="BusStopCode", min_length=5, max_length=5),
     service_no: str | None = Query(None, alias="ServiceNo"),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {"BusStopCode": bus_stop_code}
     if service_no:
         params["ServiceNo"] = service_no
     try:
-        return await client.get("/v3/BusArrival", params=params)
+        return await _fetch_with_cache("/v3/BusArrival", params, client, cache, ttl_seconds=15)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -43,6 +66,7 @@ async def bus_services(
     service_no: str | None = Query(None, alias="ServiceNo"),
     skip: int | None = Query(None, alias="$skip", ge=0),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if service_no:
@@ -50,7 +74,7 @@ async def bus_services(
     if skip is not None:
         params["$skip"] = skip
     try:
-        return await client.get("/BusServices", params=params)
+        return await _fetch_with_cache("/BusServices", params, client, cache, ttl_seconds=300)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -59,12 +83,13 @@ async def bus_services(
 async def bus_routes(
     skip: int | None = Query(None, alias="$skip", ge=0),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if skip is not None:
         params["$skip"] = skip
     try:
-        return await client.get("/BusRoutes", params=params)
+        return await _fetch_with_cache("/BusRoutes", params, client, cache, ttl_seconds=300)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -74,6 +99,7 @@ async def bus_stops(
     bus_stop_code: str | None = Query(None, alias="BusStopCode", min_length=5, max_length=5),
     skip: int | None = Query(None, alias="$skip", ge=0),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if bus_stop_code:
@@ -81,7 +107,7 @@ async def bus_stops(
     if skip is not None:
         params["$skip"] = skip
     try:
-        return await client.get("/BusStops", params=params)
+        return await _fetch_with_cache("/BusStops", params, client, cache, ttl_seconds=1800)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -90,12 +116,13 @@ async def bus_stops(
 async def passenger_volume_bus(
     date: str | None = Query(None, alias="Date", pattern=r"^[0-9]{6}$"),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if date:
         params["Date"] = date
     try:
-        return await client.get("/PV/Bus", params=params)
+        return await _fetch_with_cache("/PV/Bus", params, client, cache, ttl_seconds=21600)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -104,12 +131,13 @@ async def passenger_volume_bus(
 async def passenger_volume_od_bus(
     date: str | None = Query(None, alias="Date", pattern=r"^[0-9]{6}$"),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if date:
         params["Date"] = date
     try:
-        return await client.get("/PV/ODBus", params=params)
+        return await _fetch_with_cache("/PV/ODBus", params, client, cache, ttl_seconds=21600)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err
 
@@ -118,11 +146,12 @@ async def passenger_volume_od_bus(
 async def planned_bus_routes(
     skip: int | None = Query(None, alias="$skip", ge=0),
     client: LTAClient = Depends(get_lta_client),
+    cache: CacheClient = Depends(get_cache_client),
 ) -> dict[str, Any]:
     params: dict[str, Any] = {}
     if skip is not None:
         params["$skip"] = skip
     try:
-        return await client.get("/PlannedBusRoutes", params=params)
+        return await _fetch_with_cache("/PlannedBusRoutes", params, client, cache, ttl_seconds=900)
     except httpx.HTTPStatusError as err:
         raise _map_httpx_error(err) from err

app/core/config.py

@@ -19,6 +19,11 @@ class Settings(BaseSettings):
     max_connections: int = 200
     max_keepalive_connections: int = 100
 
+    valkey_enabled: bool = True
+    valkey_url: str = "redis://valkey:6379/0"
+    valkey_connect_timeout_seconds: float = 1.0
+    valkey_default_ttl_seconds: int = 120
+
 
 @lru_cache
 def get_settings() -> Settings:

app/main.py

@@ -4,17 +4,22 @@ from fastapi import FastAPI
 
 from app.api.routes import bus, health
 from app.core.config import get_settings
+from app.services.cache import CacheClient
 from app.services.lta_client import LTAClient
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     settings = get_settings()
-    client = LTAClient(settings)
-    await client.startup()
-    app.state.lta_client = client
+    lta_client = LTAClient(settings)
+    cache_client = CacheClient(settings)
+    await lta_client.startup()
+    await cache_client.startup()
+    app.state.lta_client = lta_client
+    app.state.cache_client = cache_client
     yield
-    await client.shutdown()
+    await lta_client.shutdown()
+    await cache_client.shutdown()
 
 
 def create_app() -> FastAPI:

app/services/cache.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from typing import Any
+
+import valkey.asyncio as valkey
+from valkey.exceptions import ValkeyError
+
+from app.core.config import Settings
+
+logger = logging.getLogger(__name__)
+
+
+class CacheClient:
+    def __init__(self, settings: Settings):
+        self._settings = settings
+        self._client: valkey.Valkey | None = None
+
+    async def startup(self) -> None:
+        if not self._settings.valkey_enabled:
+            return
+
+        self._client = valkey.from_url(
+            self._settings.valkey_url,
+            decode_responses=True,
+            socket_connect_timeout=self._settings.valkey_connect_timeout_seconds,
+        )
+        try:
+            await self._client.ping()
+        except ValkeyError:
+            logger.exception("Valkey ping failed; continuing without cache")
+            await self.shutdown()
+
+    async def shutdown(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    async def get_json(self, key: str) -> dict[str, Any] | None:
+        if self._client is None:
+            return None
+
+        try:
+            value = await self._client.get(key)
+            if value is None:
+                return None
+            return json.loads(value)
+        except (ValkeyError, json.JSONDecodeError):
+            logger.exception("Valkey get failed for key=%s", key)
+            return None
+
+    async def set_json(self, key: str, value: dict[str, Any], ttl_seconds: int | None = None) -> None:
+        if self._client is None:
+            return
+
+        ttl = ttl_seconds or self._settings.valkey_default_ttl_seconds
+        try:
+            payload = json.dumps(value, separators=(",", ":"), ensure_ascii=True)
+            await self._client.set(key, payload, ex=ttl)
+        except ValkeyError:
+            logger.exception("Valkey set failed for key=%s", key)
+
+    @staticmethod
+    def build_cache_key(path: str, params: dict[str, Any] | None = None) -> str:
+        canonical = json.dumps(params or {}, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
+        digest = hashlib.sha1(f"{path}?{canonical}".encode("utf-8")).hexdigest()
+        return f"lta:{path}:{digest}"

docker-compose.yml

@@ -1,4 +1,11 @@
 services:
+  valkey:
+    image: valkey/valkey:8-alpine
+    command: ["valkey-server", "--save", "", "--appendonly", "no"]
+    ports:
+      - "6379:6379"
+    restart: unless-stopped
+
   api:
     build:
       context: .
@@ -6,6 +13,8 @@ services:
     image: lta-datamall-bus-backend:latest
     env_file:
       - .env
+    depends_on:
+      - valkey
     ports:
       - "8000:8000"
     healthcheck:

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
   "uvicorn[standard]==0.35.0",
   "gunicorn==23.0.0",
   "httpx==0.28.1",
+  "valkey==6.1.0",
   "pydantic==2.11.7",
   "pydantic-settings==2.10.1",
 ]

uv.lock

@@ -173,6 +173,7 @@ dependencies = [
     { name = "pydantic" },
     { name = "pydantic-settings" },
     { name = "uvicorn", extra = ["standard"] },
+    { name = "valkey" },
 ]
 
 [package.metadata]
@@ -183,6 +184,7 @@ requires-dist = [
     { name = "pydantic", specifier = "==2.11.7" },
     { name = "pydantic-settings", specifier = "==2.10.1" },
     { name = "uvicorn", extras = ["standard"], specifier = "==0.35.0" },
+    { name = "valkey", specifier = "==6.1.0" },
 ]
 
 [[package]]
@@ -410,6 +412,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/e4/16/c1fd27e9549f3c4baf1dc9c20c456cd2f822dbf8de9f463824b0c0357e06/uvloop-0.22.1-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:6cde23eeda1a25c75b2e07d39970f3374105d5eafbaab2a4482be82f272d5a5e", size = 4296730, upload-time = "2025-10-16T22:17:00.744Z" },
 ]
 
+[[package]]
+name = "valkey"
+version = "6.1.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/38/d4/04b6a234e584e21ccb63b895ca9dfb4e759e4c139c1ab3f9484982ee6491/valkey-6.1.0.tar.gz", hash = "sha256:a652df15ed89c41935ffae6dfd09c56f4a9ab80b592e5ed9204d538e2ddad6d3", size = 4600944, upload-time = "2025-02-11T17:20:46.706Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/79/b0/c4d47032bbda89cff7af99c0b096db9b9453b9f0c1e24cf027aa616be389/valkey-6.1.0-py3-none-any.whl", hash = "sha256:cfe769edae894f74ac946eff1e93f7d7f466032c3030ba7e9d089a742459ac9c", size = 259302, upload-time = "2025-02-11T17:20:42.96Z" },
+]
+
 [[package]]
 name = "watchfiles"
 version = "1.2.0"