From f10ab4e7b33fcf053dfcdf8a1917f8ee0a91bf4c Mon Sep 17 00:00:00 2001 From: furyhawk Date: Sun, 31 May 2026 10:53:14 +0800 Subject: [PATCH] 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 | 4 +++ AGENTS.md | 2 ++ Makefile | 12 +++++++- README.md | 22 ++++++++++++++ app/api/routes/bus.py | 43 ++++++++++++++++++++++----- app/core/config.py | 5 ++++ app/main.py | 13 +++++--- app/services/cache.py | 69 +++++++++++++++++++++++++++++++++++++++++++ docker-compose.yml | 9 ++++++ pyproject.toml | 1 + uv.lock | 11 +++++++ 11 files changed, 179 insertions(+), 12 deletions(-) create mode 100644 app/services/cache.py diff --git a/.env.example b/.env.example index 39961d7..06175de 100644 --- a/.env.example +++ b/.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 diff --git a/AGENTS.md b/AGENTS.md index 805e674..03a0ae4 100644 --- a/AGENTS.md +++ b/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). diff --git a/Makefile b/Makefile index fa7862b..754c14b 100644 --- a/Makefile +++ b/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 diff --git a/README.md b/README.md index f19e4c1..f25513d 100644 --- a/README.md +++ b/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 diff --git a/app/api/routes/bus.py b/app/api/routes/bus.py index 0a6e38a..7eaacf4 100644 --- a/app/api/routes/bus.py +++ b/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 diff --git a/app/core/config.py b/app/core/config.py index 18e1615..4868b47 100644 --- a/app/core/config.py +++ b/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: diff --git a/app/main.py b/app/main.py index e8ba0f8..3615f6b 100644 --- a/app/main.py +++ b/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: diff --git a/app/services/cache.py b/app/services/cache.py new file mode 100644 index 0000000..e749fd5 --- /dev/null +++ b/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}" diff --git a/docker-compose.yml b/docker-compose.yml index fca205f..bc0831d 100644 --- a/docker-compose.yml +++ b/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: diff --git a/pyproject.toml b/pyproject.toml index f346c3a..b4d9562 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/uv.lock b/uv.lock index 72dfbfe..32aa59d 100644 --- a/uv.lock +++ b/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"