This is an example how to by pass cache for a specific request in dogpile cache library. The code is pretty self-explanatory. The actual bypass happens in the `BypassProxy` class

flask_dogpile_cache_bypass.py

# /// script
# requires-python = ">=3.13"
# dependencies = [
#   "dogpile.cache",
#   "flask",
#   "loguru",
# ]
# ///

from __future__ import annotations
from dogpile.cache.api import NO_VALUE
from dogpile.cache import make_region, CacheRegion
from dogpile.cache.proxy import ProxyBackend
from flask import Flask, request, g
from loguru import logger


DOGPILE_CACHE = "dogpile.cache.memory"
DOGPILE_CACHE = "dogpile.cache.memory_pickle"
TIMEOUT_SECONDS = 2 * 60 * 60

app = Flask(__name__)


def str2bool(v: str):
    if v.lower() in ("yes", "true", "t", "y", "1"):
        return True
    elif v.lower() in ("no", "false", "f", "n", "0"):
        return False


def key_generator(namespace, fn):
    fname = fn.__name__

    def generate_key(*arg):
        key_template = f"{namespace}_{fname}_{'_'.join(str(s) for s in arg)}"
        return key_template

    return generate_key


class BypassProxy(ProxyBackend):
    def get(self, key):
        if g.cache_bypass:
            logger.warning(f"Bypassing cache: {key}")
            return NO_VALUE

        return self.proxied.get(key)

    def get_multi(self, keys):
        if g.cache_bypass:
            logger.warning(f"Bypassing cache: {keys}")
            return [NO_VALUE for _ in keys]

        return self.proxied.get_multi(keys)

    def get_serialized(self, key):
        if g.cache_bypass:
            logger.warning(f"Bypassing cache: {key}")
            return NO_VALUE

        logger.info(f"Getting Cache Serialized Key: {key}")
        try:
            return self.proxied.get_serialized(key)
        except Exception:
            return

    def get_serialized_multi(self, keys):
        if g.cache_bypass:
            logger.warning(f"Bypassing cache: {keys}")
            return [NO_VALUE for _ in keys]

        logger.info(f"Getting Cache Serialized Keys: {keys}")
        return self.proxied.get_serialized_multi(keys)


cache_region: CacheRegion = make_region(
    key_mangler=lambda key: f"rest-api:dogpile:{key}",
).configure(
    DOGPILE_CACHE,
    expiration_time=TIMEOUT_SECONDS,
    wrap=[
        BypassProxy,
    ],
)


_global_users = set()


@cache_region.cache_on_arguments(namespace="users", function_key_generator=key_generator)
def get_users():
    logger.info("Computing fresh")
    return _global_users


@app.get(
    "/",
)
def return_users():
    logger.info("Return users")
    g.cache_bypass = str2bool(request.headers.get("x-bypass-cache", "False"))
    mus = get_users()
    return f"""{",".join(mus)}""", 200


@app.post(
    "/<i>",
)
def create_user(i: int):
    logger.info("Create user")
    _global_users.add(i)
    return f"Created user: {i}", 200


def main():
    app.run(
        host="0.0.0.0",
        port=2000,
        debug=True,
        threaded=True,
    )


if __name__ == "__main__":
    main()