Python Storyblok Cache Demo: demonstrates the difference between cached and non-cached Content Delivery API requests.

storyblok_cache_demo.py

#!/usr/bin/env python3
"""
Python Storyblok Cache Demo
Demonstrates the difference between cached and non-cached Content Delivery API requests.

Based on: https://www.storyblok.com/faq/how-stories-are-cached-content-delivery-api
You have to create a .env with
STORYBLOK_TOKEN=your-space-access-token

Install 2 packages:
pip3 install python-dotenv
pip3 install requests
"""

import os
import time
import requests
from dataclasses import dataclass
from dotenv import load_dotenv


@dataclass
class RequestResult:
    """Result of an API request with timing info."""
    status_code: int
    elapsed_ms: float
    cv: int | None = None
    data: dict | None = None


class StoryblokCacheDemo:
    """Demonstrates Storyblok Content Delivery API caching behavior."""

    BASE_URL = "https://api.storyblok.com/v2/cdn"

    def __init__(self, token: str):
        self.token = token
        self.cached_cv: int | None = None

    def _request(self, endpoint: str, params: dict | None = None) -> RequestResult:
        """Make a timed request to the Storyblok API."""
        url = f"{self.BASE_URL}/{endpoint}"
        request_params = {"token": self.token, "version": "published"}
        if params:
            request_params.update(params)

        start_time = time.perf_counter()
        first_response = requests.get(url, params=request_params, allow_redirects=False)

        if first_response.status_code == 301:
            redirect_url = first_response.headers.get('Location')
            print(f"   Initial: {first_response.status_code} from {first_response.url}")
            print(f"   X-Cache: {first_response.headers.get('x-cache', 'N/A')}")
            final_response = requests.get(redirect_url)
            print(f"   Final: {final_response.status_code} from {final_response.url}")
            print(f"   X-Cache: {final_response.headers.get('x-cache', 'N/A')}")
            response = final_response
        else:
            print(f"   Response: {first_response.status_code} from {first_response.url}")
            print(f"   X-Cache: {first_response.headers.get('x-cache', 'N/A')}")
            response = first_response
        elapsed_ms = (time.perf_counter() - start_time) * 1000

        data = response.json() if response.status_code == 200 else None
        cv = request_params.get('cv', 'N/A')

        return RequestResult(
            status_code=response.status_code,
            elapsed_ms=elapsed_ms,
            cv=cv,
            data=data,
        )

    def get_space_info(self) -> RequestResult:
        """Get space info including current cache version."""
        result = self._request("spaces/me")
        if result.data and "space" in result.data:
            self.cached_cv = result.data["space"].get("version")
            result.cv = self.cached_cv
        return result


    def get_story_no_cache(self, slug: str, current_timestamp: int | None = None) -> RequestResult:
        """Fetch a specific story bypassing CDN cache."""
        return self._request(f"stories/{slug}", {"cv": current_timestamp})

    def get_story_with_cache(self, slug: str, cv: int | None = None) -> RequestResult:
        """Fetch a specific story using CDN cache."""
        version = cv or self.cached_cv
        params = {"cv": version} if version else {}
        result = self._request(f"stories/{slug}", params)
        if result.cv and not self.cached_cv:
            self.cached_cv = result.cv
        return result

    def flush_cache(self) -> None:
        """Clear the locally stored cache version."""
        self.cached_cv = None
        print("🗑️  Local cache version cleared.")


def print_result(label: str, result: RequestResult) -> None:
    """Print formatted request result."""
    status_icon = "✅" if result.status_code == 200 else "❌"
    print(f"{status_icon} {label}")
    print(f"   Status: {result.status_code}")
    print(f"   Time:   {result.elapsed_ms:.2f} ms")
    if result.cv:
        print(f"   CV:     {result.cv}")
    print()


def run_demo(token: str) -> None:
    """Run the cache demonstration."""
    demo = StoryblokCacheDemo(token)

    print("=" * 60)
    print("🚀 Storyblok Content Delivery API - Cache Demo")
    print("=" * 60)
    print()

    # Step 1: Get space info to retrieve current cache version
    print("📋 Step 1: Fetching space info to get current cache version...")
    result = demo.get_space_info()
    print_result("Space Info", result)

    if result.status_code != 200:
        print("❌ Failed to connect. Check your token.")
        return

    cv = demo.cached_cv
    print(f"📌 Current cache version (cv): {cv}")
    print()

    # Step 2: Fetch stories WITHOUT cache (bypass CDN)
    print("-" * 60)
    print("📋 Step 2: Fetching stories WITHOUT cache (bypass CDN)...")
    print("   Using cv=Date.now() to force fresh content")
    print()

    no_cache_times = []
    for i in range(3):

        current_timestamp = int(time.time() * 1000)  # milliseconds
        result = demo.get_story_no_cache("about", current_timestamp)
        no_cache_times.append(result.elapsed_ms)
        print_result(f"Request {i + 1} (no cache)", result)
        time.sleep(0.1)  # Small delay between requests

    # Step 3: Fetch stories WITH cache (use CDN)
    print("-" * 60)
    print("📋 Step 3: Fetching stories WITH cache (use CDN)...")
    print(f"   Using cv={cv} to leverage CDN cache")
    print()

    cached_times = []
    for i in range(3):
        result = demo.get_story_with_cache("about", cv)
        cached_times.append(result.elapsed_ms)
        print_result(f"Request {i + 1} (cached)", result)
        time.sleep(0.1)

    # Step 4: Compare results
    print("=" * 60)
    print("📊 RESULTS COMPARISON")
    print("=" * 60)
    print()

    avg_no_cache = sum(no_cache_times) / len(no_cache_times)
    avg_cached = sum(cached_times) / len(cached_times)

    print(f"🔴 Without cache (bypass CDN):")
    print(f"   Requests: {', '.join(f'{t:.2f}ms' for t in no_cache_times)}")
    print(f"   Average:  {avg_no_cache:.2f} ms")
    print()

    print(f"🟢 With cache (CDN):")
    print(f"   Requests: {', '.join(f'{t:.2f}ms' for t in cached_times)}")
    print(f"   Average:  {avg_cached:.2f} ms")
    print()

    if avg_cached < avg_no_cache:
        speedup = ((avg_no_cache - avg_cached) / avg_no_cache) * 100
        print(f"⚡ Cache speedup: {speedup:.1f}% faster")
        print(f"   Saved: {avg_no_cache - avg_cached:.2f} ms per request")
    else:
        print("ℹ️  No significant speedup observed (network variance)")

    print()
    print("💡 Tips:")
    print("   - Cached requests benfits of a higher rate limits see: https://www.storyblok.com/docs/api/content-delivery/v2/getting-started/rate-limit")
    print("   - Store cv value and reuse it for subsequent requests")
    print("   - Use webhooks to invalidate cache when content changes")
    print()


def main() -> None:
    """Main entry point."""
    load_dotenv()  # Loads variables from .env file
    token = os.getenv("STORYBLOK_TOKEN")


    if not token:
        print("❌ Error: STORYBLOK_TOKEN environment variable not set.")
        print()
        print("Usage:")
        print("  export STORYBLOK_TOKEN='your_preview_or_public_token'")
        print("  python storyblok_cache_demo.py")
        print()
        print("You can find your token in Storyblok:")
        print("  Settings > Access Tokens")
        return

    run_demo(token)


if __name__ == "__main__":
    main()