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 fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
javascript 
import { useState, useEffect, useCallback, useRef } from 'react';
 
const cache = new Map();
const inflight = new Map();

Building a stale-while-revalidate hook in React

caching request-deduplication custom-hooks
Advanced 10 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
php 
<?php
 
namespace App\Services;
 

Building a cached daily leaderboard in Laravel

caching aggregation eager-loading
Intermediate 9 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 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