How I Cut My API Costs by 60% With a 50-Line Python Cache

Last month I got a bill from an AI API provider that made me choke on my coffee. $340. For a side project that maybe 50 people use.

The culprit? I was calling the same endpoints with the same parameters over and over. Every user refresh hit the API. Every page reload. Every test run.

I checked the logs and found something embarrassing: 73% of my API calls were identical duplicates — same endpoint, same payload, same response.

That's when I built this. It took 30 minutes and I haven't touched it since. Here's the code, why it works, and the one edge case that almost broke everything.

The Problem, Quantified

Before I show you the code, here's what my API usage looked like:

Metric	Before	After
Daily API calls	8,400	2,300
Duplicate call rate	73%	12%
Monthly bill	$340	$135
Avg response time	1.2s	80ms

The 12% that still slip through are dynamic queries — things like "what's the weather right now" where caching makes no sense. Everything else gets served from disk.

The Code

Here's the full thing. 50 lines of Python, no dependencies outside the standard library:

import hashlib
import json
import os
import time
from functools import wraps

CACHE_DIR = "./api_cache"
DEFAULT_TTL = 3600  # 1 hour


def _make_key(method: str, url: str, params: dict, body: dict) -> str:
    """Create a deterministic cache key from request components."""
    payload = json.dumps({
        "m": method,
        "u": url,
        "p": sorted(params.items()) if params else [],
        "b": sorted(body.items()) if body else [],
    }, sort_keys=True)
    return hashlib.sha256(payload.encode()).hexdigest()


def _path(key: str) -> str:
    return os.path.join(CACHE_DIR, key[:2], key)


def api_cache(ttl: int = DEFAULT_TTL):
    """Cache API responses to disk with configurable TTL."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Extract identifiable request components
            # Adjust these based on how YOUR API client works
            req = kwargs.get("request_data", {})
            key = _make_key(
                method=req.get("method", "GET"),
                url=req.get("url", func.__name__),
                params=req.get("params", {}),
                body=req.get("body", {}),
            )

            cache_path = _path(key)

            # Cache hit — check freshness
            if os.path.exists(cache_path):
                age = time.time() - os.path.getmtime(cache_path)
                if age < ttl:
                    with open(cache_path, "r") as f:
                        return json.load(f)

            # Cache miss — call the real API
            result = func(*args, **kwargs)

            # Don't cache errors. This is the part I got wrong the first time.
            if result and not isinstance(result, Exception):
                os.makedirs(os.path.dirname(cache_path), exist_ok=True)
                with open(cache_path, "w") as f:
                    json.dump(result, f)

            return result

        return wrapper
    return decorator

How to Use It

Wrap your API call function and pass in request metadata:

import requests

@api_cache(ttl=1800)  # 30 minutes
def call_openai(prompt: str, request_data: dict = None):
    resp = requests.post(
        "https://api.openai.com/v1/chat/completions",
        headers={"Authorization": f"Bearer {os.environ['OPENAI_KEY']}"},
        json={
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": prompt}],
        },
    )
    return resp.json()


# Call it — first time hits the API, subsequent calls in the
# next 30 minutes return from cache
result = call_openai(
    "summarize this text: ...",
    request_data={
        "method": "POST",
        "url": "openai/chat/completions",
        "params": {},
        "body": {"prompt": "summarize this text: ..."},
    },
)

For more complex APIs, include the relevant distinguishing fields in request_data.body. The hash function handles the rest.

The Edge Case I Got Wrong

On day one, I cached everything. Including errors.

My AI provider went down for 8 minutes. During that window, my cache stored 429 responses — HTTP errors — and served them to every user for the full TTL. Users saw "rate limit exceeded" for things that normally worked fine.

That's why there's this check in the code:

if result and not isinstance(result, Exception):
    os.makedirs(os.path.dirname(cache_path), exist_ok=True)
    ...

Two rules I now follow:

1. Never cache 4xx or 5xx responses.
2. Never cache empty responses. They hide real outages.

If you're wrapping requests, here's a quick filter you can add:

def _is_cacheable(response) -> bool:
    if response is None:
        return False
    if isinstance(response, dict) and "error" in response:
        return False
    # If using requests.Response object
    if hasattr(response, "status_code") and response.status_code >= 400:
        return False
    return True

When NOT to Use This

This works well for:

• AI text generation with deterministic prompts (same prompt = same response)
• REST API read endpoints (GET /users/42)
• Configuration fetches
• Anything where staleness is acceptable for a few minutes

It does not work for:

• Real-time data (stock prices, live sports scores)
• Write operations (POST that creates resources)
• User-specific data unless you include user ID in the cache key
• Endpoints with rapidly changing state

I learned that last one the hard way when I cached a "current server status" endpoint and spent an afternoon debugging why my dashboard showed "healthy" for a server that had been down for two hours.

Going Further: Memory + Disk

The version above uses disk only. That's fine for hundreds of requests per minute. If you need more speed, add an in-memory LRU layer first:

from collections import OrderedDict

_mem_cache: OrderedDict[str, tuple[float, dict]] = OrderedDict()
MAX_MEM_ITEMS = 500

def _cache_get(key: str, ttl: int) -> dict | None:
    # Check memory first
    if key in _mem_cache:
        ts, val = _mem_cache[key]
        if time.time() - ts < ttl:
            _mem_cache.move_to_end(key)
            return val
        del _mem_cache[key]

    # Fall back to disk
    cache_path = _path(key)
    if os.path.exists(cache_path):
        age = time.time() - os.path.getmtime(cache_path)
        if age < ttl:
            with open(cache_path) as f:
                val = json.load(f)
            _mem_cache[key] = (time.time(), val)
            if len(_mem_cache) > MAX_MEM_ITEMS:
                _mem_cache.popitem(last=False)
            return val

    return None

This adds about 15 lines and makes cache hits virtually instant. For my use case the disk-only version was enough, but if you're serving web traffic, the memory layer is worth it.

What I'd Do Differently

If I were building this again from scratch:

1. Use SQLite instead of files. Better concurrent access, built-in TTL with a timestamp column, and one file instead of thousands.
2. Add cache statistics. Hit rate, miss rate, total bytes saved. I ended up adding this later anyway to justify the build to myself.
3. Make TTL configurable per endpoint. Some endpoints are fine at 1 hour, others need 5 minutes.

But honestly? The 50-line version has been running untouched for a month and saved me $205. Sometimes the simple thing is good enough.

---

Have you built something similar? What's your approach to API cost control? I'd love to hear about it in the comments.

Comments

[TokenLat]

Nice — this is the app-layer win most people skip. The gateway-level complement: provider-native prompt caching. The silent tax is non-deterministic prompt formatting (timestamps, key ordering) that quietly kills cache reuse even when the call is semantically identical. Normalizing prompts before they hit cache — and pinning a provider per request so cache affinity holds — is what turns a 60% win into a sustained one. And yes: never cache 4xx/5xx.