python 47 lines · 7 steps

Custom cache keys for Flask endpoints

A Flask analytics blueprint caches expensive revenue reports in Redis using a query-aware cache key, then invalidates it on demand.

Explained by highlit
Flask Intermediate caching redis cache-invalidation blueprints rest-api
1from flask import Blueprint, jsonify, request, abort
2from flask_caching import Cache
3 
4cache = Cache(config={"CACHE_TYPE": "RedisCache", "CACHE_REDIS_URL": "redis://localhost:6379/0"})
5 
6analytics = Blueprint("analytics", __name__, url_prefix="/analytics")
7 
8 
9def _report_cache_key():
10 org_id = request.view_args["org_id"]
11 granularity = request.args.get("granularity", "daily")
12 window = request.args.get("window", "30d")
13 return f"report:{org_id}:{granularity}:{window}"
14 
15 
16@analytics.route("/orgs/<int:org_id>/report")
17@cache.cached(timeout=900, make_cache_key=_report_cache_key)
18def revenue_report(org_id):
19 granularity = request.args.get("granularity", "daily")
20 window = request.args.get("window", "30d")
21 
22 if granularity not in {"daily", "weekly", "monthly"}:
23 abort(400, description="unsupported granularity")
24 
25 rows = ReportingService.aggregate_revenue(
26 org_id=org_id,
27 granularity=granularity,
28 window=window,
29 )
30 
31 return jsonify(
32 {
33 "org_id": org_id,
34 "granularity": granularity,
35 "window": window,
36 "series": [row.as_dict() for row in rows],
37 "total": sum(row.amount for row in rows),
38 }
39 )
40 
41 
42@analytics.route("/orgs/<int:org_id>/report/invalidate", methods=["POST"])
43def invalidate_report(org_id):
44 for granularity in ("daily", "weekly", "monthly"):
45 for window in ("7d", "30d", "90d"):
46 cache.delete(f"report:{org_id}:{granularity}:{window}")
47 return "", 204
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A custom make_cache_key lets you vary cached responses by query params, not just the URL path.
  2. 2The cache key format is a contract shared between the read path and the invalidation path.
  3. 3Validate inputs before doing expensive work, and let a POST endpoint explicitly purge stale entries.

Related explainers

python 
from flask import Blueprint, jsonify, request, abort
 
v1 = Blueprint("users_v1", __name__)
v2 = Blueprint("users_v2", __name__)

Versioning a Flask API with Blueprints

api versioning blueprints rest
Intermediate 7 steps
python 
import time
import threading
from enum import Enum
from functools import wraps

Building a circuit breaker in Python

circuit-breaker resilience decorators
Advanced 7 steps
php 
<?php
 
namespace App\Http\Middleware;
 

Idempotent requests with Laravel middleware

idempotency middleware caching
Advanced 8 steps
python 
from django.contrib.postgres.search import SearchQuery, SearchRank, SearchVector
from django.core.cache import cache
from django.db.models import F
from django.http import JsonResponse

Ranked full-text search in Django

full-text-search debouncing caching
Advanced 9 steps
python 
import os
from datetime import timedelta
 
 

Class-based config in a Flask app factory

configuration app-factory inheritance
Intermediate 7 steps
python 
import time
 
from flask import Flask, g, request
 

Timing requests with Flask hooks

middleware request-lifecycle instrumentation
Intermediate 5 steps

Share this explainer

Here's the card — post it anywhere.

Custom cache keys for Flask endpoints — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code