Caching

from flask import Flask, jsonify

app = Flask("bookshelf")

BOOKS = [{"id": 1, "title": "Dune"}, {"id": 2, "title": "Hyperion"}]


@app.get("/books")
def list_books():
    response = jsonify(BOOKS)
    response.cache_control.public = True
    response.cache_control.max_age = 60
    response.cache_control.s_maxage = 300
    return response


@app.get("/account")
def account():
    response = jsonify({"user": "reader"})
    response.cache_control.no_store = True
    return response


client = app.test_client()
cache_control = client.get("/books").headers["Cache-Control"]
assert set(cache_control.split(", ")) == {"public", "max-age=60", "s-maxage=300"}
assert client.get("/account").headers["Cache-Control"] == "no-store"

from flask import Flask, jsonify, request

app = Flask("bookshelf-vary")


@app.get("/greeting")
def greeting():
    language = request.accept_languages.best_match(["en", "es"], default="en")
    response = jsonify({"language": language})
    response.vary.add("Accept-Language")
    response.cache_control.public = True
    response.cache_control.max_age = 600
    return response


client = app.test_client()
response = client.get("/greeting", headers={"Accept-Language": "es"})
assert response.json == {"language": "es"}
assert "Accept-Language" in response.headers["Vary"]

from flask import Flask, jsonify, request

app = Flask("bookshelf-etag")

STATE = {"version": 7, "books": [{"id": 1, "title": "Dune"}]}
WORK_DONE = []


@app.get("/books")
def list_books():
    etag = f"books-{STATE['version']}"
    if etag in request.if_none_match:
        response = app.response_class(status=304)
        response.set_etag(etag)
        return response
    WORK_DONE.append(etag)  # the expensive query would run here
    response = jsonify(STATE["books"])
    response.set_etag(etag)
    response.cache_control.no_cache = True
    return response


client = app.test_client()
first = client.get("/books")
assert first.status_code == 200
etag = first.headers["ETag"]
assert etag == '"books-7"'

second = client.get("/books", headers={"If-None-Match": etag})
assert second.status_code == 304 and second.data == b""
assert len(WORK_DONE) == 1  # the second request did no work

STATE["version"] = 8
third = client.get("/books", headers={"If-None-Match": etag})
assert third.status_code == 200 and third.headers["ETag"] == '"books-8"'

from flask import Flask, jsonify, request

app = Flask("bookshelf-conditional")


@app.get("/books")
def list_books():
    response = jsonify([{"id": 1, "title": "Dune"}])
    response.add_etag()
    return response.make_conditional(request)


client = app.test_client()
first = client.get("/books")
assert first.status_code == 200
validator = first.headers["ETag"]
again = client.get("/books", headers={"If-None-Match": validator})
assert again.status_code == 304

from flask import Flask
from flask_caching import Cache

cache = Cache()  # unbound, created at import time


def create_app():
    app = Flask("bookshelf-cache")
    app.config.from_mapping(
        CACHE_TYPE="SimpleCache",
        CACHE_DEFAULT_TIMEOUT=120,
        CACHE_THRESHOLD=200,
        CACHE_KEY_PREFIX="bookshelf_",
    )
    cache.init_app(app)
    return app


app = create_app()
with app.app_context():
    assert cache.set("greeting", "hello", timeout=30)
    assert cache.get("greeting") == "hello"
    assert cache.get("missing") is None
    assert cache.add("greeting", "ignored") is False  # add only stores a new key
    cache.delete("greeting")
    assert cache.get("greeting") is None
    cache.set_many({"a": 1, "b": 2})
    assert cache.get_many("a", "b") == [1, 2]
    assert cache.get_dict("a", "b") == {"a": 1, "b": 2}
    cache.delete_many("a", "b")
    assert cache.has("a") is False
    cache.clear()

# norun
import os

from flask import Flask
from flask_caching import Cache

app = Flask("bookshelf-redis")
app.config.from_mapping(
    CACHE_TYPE="RedisCache",
    CACHE_REDIS_URL=os.environ["BOOKSHELF_REDIS_URL"],  # e.g. rediss://host:6379/0, with credentials from a secrets store
    CACHE_DEFAULT_TIMEOUT=300,
    CACHE_KEY_PREFIX="bookshelf:prod:",
)
cache = Cache(app)

import fakeredis
from flask import Flask
from flask_caching import Cache
from flask_caching.backends import RedisCache

app = Flask("bookshelf-fakeredis")
app.config["CACHE_TYPE"] = "SimpleCache"
cache = Cache(app)

# In a test, swap the backend for one that talks to an in-process fake Redis.
fake = fakeredis.FakeRedis()
app.extensions["cache"][cache] = RedisCache(host=fake, key_prefix="bookshelf:test:", default_timeout=60)

with app.app_context():
    cache.set("book:1", {"title": "Dune"})
    assert cache.get("book:1") == {"title": "Dune"}
assert fake.exists("bookshelf:test:book:1") == 1
assert 0 < fake.ttl("bookshelf:test:book:1") <= 60

from flask import Flask, jsonify, request
from flask_caching import Cache

app = Flask("bookshelf-cached")
app.config.from_mapping(CACHE_TYPE="SimpleCache", CACHE_DEFAULT_TIMEOUT=60)
cache = Cache(app)

CALLS = {"count": 0}


def expensive_catalogue():
    CALLS["count"] += 1
    return [{"id": 1, "title": "Dune"}, {"id": 2, "title": "Hyperion"}]


@app.get("/catalogue")
@cache.cached(timeout=120)
def catalogue():
    return jsonify(expensive_catalogue())


@app.get("/search")
@cache.cached(timeout=120, query_string=True)
def search():
    term = request.args.get("q", "")
    CALLS["count"] += 1
    return jsonify({"q": term})


client = app.test_client()
client.get("/catalogue")
client.get("/catalogue")
assert CALLS["count"] == 1  # the second call came from the cache

client.get("/search?q=dune&page=1")
client.get("/search?page=1&q=dune")  # same query, different order: same key
assert CALLS["count"] == 2
client.get("/search?q=hyperion")
assert CALLS["count"] == 3

from flask import Flask, abort, jsonify, request
from flask_caching import Cache

app = Flask("bookshelf-cached-options")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

CALLS = []


def is_admin_request():
    return request.headers.get("X-Role") == "admin"


@app.get("/stats")
@cache.cached(timeout=300, unless=is_admin_request, response_hit_indication=True)
def stats():
    CALLS.append(1)
    return jsonify({"books": 2})


@app.get("/books/<int:book_id>")
@cache.cached(timeout=300, response_filter=lambda rv: getattr(rv, "status_code", 200) == 200)
def book(book_id):
    if book_id != 1:
        abort(404)
    return jsonify({"id": 1})


client = app.test_client()
client.get("/stats")
hit = client.get("/stats")
assert len(CALLS) == 1 and "hit_cache" in hit.headers

client.get("/stats", headers={"X-Role": "admin"})  # bypasses the cache entirely
assert len(CALLS) == 2

assert client.get("/books/1").status_code == 200
assert client.get("/books/9").status_code == 404

assert stats.uncached.__name__ == "stats"  # the undecorated function

from flask import Flask, g, jsonify
from flask_caching import Cache

app = Flask("bookshelf-per-user")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

CALLS = []


@app.before_request
def load_user():
    from flask import request

    g.user_id = request.headers.get("X-User", "anonymous")


def per_user_key():
    from flask import request

    return f"shelf/{g.user_id}{request.path}"


@app.get("/shelf")
@cache.cached(timeout=60, key_prefix=per_user_key)
def shelf():
    CALLS.append(g.user_id)
    return jsonify({"owner": g.user_id})


client = app.test_client()
assert client.get("/shelf", headers={"X-User": "ana"}).json == {"owner": "ana"}
assert client.get("/shelf", headers={"X-User": "ben"}).json == {"owner": "ben"}
assert client.get("/shelf", headers={"X-User": "ana"}).json == {"owner": "ana"}
assert CALLS == ["ana", "ben"]  # the third call hit the cache

from flask import Flask
from flask_caching import Cache

app = Flask("bookshelf-memoize")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

CALLS = []


@cache.memoize(timeout=300)
def books_by_author(author, limit=10):
    CALLS.append((author, limit))
    return [f"{author} book {n}" for n in range(limit)]


with app.app_context():
    first = books_by_author("Herbert", limit=2)
    again = books_by_author("Herbert", limit=2)
    assert first == again == ["Herbert book 0", "Herbert book 1"]
    assert len(CALLS) == 1

    books_by_author("Herbert", limit=3)  # different arguments, different entry
    books_by_author("Simmons", limit=2)
    assert len(CALLS) == 3

    # Invalidate exactly one argument combination ...
    cache.delete_memoized(books_by_author, "Herbert", limit=2)
    books_by_author("Herbert", limit=2)
    assert len(CALLS) == 4

    # ... or every cached result of the function.
    cache.delete_memoized(books_by_author)
    books_by_author("Simmons", limit=2)
    assert len(CALLS) == 5

from flask import Flask
from flask_caching import Cache

app = Flask("bookshelf-memoize-ignore")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

CALLS = []


@cache.memoize(60, args_to_ignore=["trace_id"])
def book_count(genre, trace_id=None):
    CALLS.append(genre)
    return 42


with app.app_context():
    book_count("sci-fi", trace_id="a1")
    book_count("sci-fi", trace_id="b2")  # same key: trace_id is ignored
    assert CALLS == ["sci-fi"]

from flask import Flask, render_template_string
from flask_caching import Cache
from flask_caching import make_template_fragment_key

app = Flask("bookshelf-fragment")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

RENDERED = []


def popular(genre):
    RENDERED.append(genre)
    return f"popular {genre} books"


PAGE = (
    "{% cache 300, 'popular', genre %}"
    "{{ popular(genre) }}"
    "{% endcache %}"
)


@app.get("/home/<genre>")
def home(genre):
    return render_template_string(PAGE, genre=genre, popular=popular)


client = app.test_client()
assert client.get("/home/scifi").get_data(as_text=True) == "popular scifi books"
client.get("/home/scifi")
assert RENDERED == ["scifi"]  # the second render came from the cache
client.get("/home/fantasy")
assert RENDERED == ["scifi", "fantasy"]  # a different vary-on value, a different entry

# Invalidate one fragment from Python by rebuilding its key.
key = make_template_fragment_key("popular", vary_on=["scifi"])
with app.app_context():
    assert cache.get(key) == "popular scifi books"
    cache.delete(key)
client.get("/home/scifi")
assert RENDERED == ["scifi", "fantasy", "scifi"]

from cachelib import SimpleCache

store = SimpleCache(threshold=100, default_timeout=30)
store.set("isbn:1", "Dune")
assert store.get("isbn:1") == "Dune"
assert store.inc("hits") == 1 and store.inc("hits") == 2
store.delete("isbn:1")
assert store.get("isbn:1") is None

from flask import Flask, jsonify, request
from flask_caching import Cache

app = Flask("bookshelf-invalidate")
app.config.from_mapping(CACHE_TYPE="SimpleCache", CACHE_DEFAULT_TIMEOUT=300)
cache = Cache(app)

DB = {1: {"id": 1, "title": "Dune"}}
QUERIES = []


@cache.memoize()
def load_book(book_id):
    QUERIES.append(book_id)
    return DB[book_id]


@app.get("/books/<int:book_id>")
def get_book(book_id):
    return jsonify(load_book(book_id))


@app.get("/books")
@cache.cached(query_string=True)
def list_books():
    return jsonify(sorted(DB.values(), key=lambda b: b["id"]))


@app.put("/books/<int:book_id>")
def update_book(book_id):
    DB[book_id] = {"id": book_id, "title": request.get_json()["title"]}
    # The "commit" succeeded: now invalidate what depends on it.
    cache.delete_memoized(load_book, book_id)
    cache.delete_cached(list_books, "/books")
    return jsonify(DB[book_id])


client = app.test_client()
assert client.get("/books/1").json["title"] == "Dune"
assert client.get("/books/1").json["title"] == "Dune"
assert QUERIES == [1]
assert client.get("/books").json[0]["title"] == "Dune"

client.put("/books/1", json={"title": "Dune Messiah"})
assert client.get("/books/1").json["title"] == "Dune Messiah"
assert client.get("/books").json[0]["title"] == "Dune Messiah"
assert QUERIES == [1, 1]

from flask import Flask
from flask_caching import Cache

app = Flask("bookshelf-version")
app.config.from_mapping(CACHE_TYPE="SimpleCache")
cache = Cache(app)

COMPUTED = []


def catalogue_version():
    return cache.get("catalogue:version") or 1


def bump_catalogue():
    cache.set("catalogue:version", catalogue_version() + 1, timeout=0)


def books_for(genre):
    key = f"catalogue:v{catalogue_version()}:{genre}"
    value = cache.get(key)
    if value is None:
        COMPUTED.append(key)
        value = [f"{genre} book"]
        cache.set(key, value, timeout=300)
    return value


with app.app_context():
    books_for("scifi")
    books_for("scifi")
    assert COMPUTED == ["catalogue:v1:scifi"]
    bump_catalogue()  # one write invalidates every genre at once
    books_for("scifi")
    assert COMPUTED == ["catalogue:v1:scifi", "catalogue:v2:scifi"]

from flask import Flask, jsonify
from flask_caching import Cache
from flask_caching.signals import cache_view_hit, cache_view_miss

app = Flask("bookshelf-signals")
app.config.from_mapping(CACHE_TYPE="SimpleCache", CACHE_ENABLE_SIGNALS=True)
cache = Cache(app)

counts = {"hit": 0, "miss": 0}


def on_hit(sender, **extra):
    counts["hit"] += 1


def on_miss(sender, **extra):
    counts["miss"] += 1


cache_view_hit.connect(on_hit, weak=False)
cache_view_miss.connect(on_miss, weak=False)


@app.get("/top")
@cache.cached(timeout=60)
def top():
    return jsonify(["Dune"])


client = app.test_client()
client.get("/top")
client.get("/top")
client.get("/top")
assert counts == {"hit": 2, "miss": 1}

import pytest
from flask import Flask, jsonify
from flask_caching import Cache

cache = Cache()


def create_app(test_config=None):
    app = Flask("bookshelf-testing")
    app.config.from_mapping(CACHE_TYPE="SimpleCache")
    if test_config:
        app.config.update(test_config)
    cache.init_app(app)

    @app.get("/n")
    @cache.cached(timeout=60)
    def counter():
        app.config["CALLS"] = app.config.get("CALLS", 0) + 1
        return jsonify(app.config["CALLS"])

    return app


def test_cache_hit():
    app = create_app({"TESTING": True})
    client = app.test_client()
    assert client.get("/n").json == client.get("/n").json == 1


def test_no_cache_in_tests():
    app = create_app({"TESTING": True, "CACHE_TYPE": "NullCache", "CACHE_NO_NULL_WARNING": True})
    client = app.test_client()
    assert client.get("/n").json == 1
    assert client.get("/n").json == 2  # nothing was cached


test_cache_hit()
test_no_cache_in_tests()