Cache Flask with Redis

from flask_alembic import Alembic
from flask_caching import Cache
from flask_sqlalchemy_lite import SQLAlchemy

from app.models import Model

db = SQLAlchemy()
alembic = Alembic(metadatas=Model.metadata)
cache = Cache()

import os
from datetime import datetime, timezone

import sqlalchemy as sa
from flask import Flask, make_response, request

from app.extensions import alembic, cache, db
from app.models import Item


def database_url() -> str:
    url = os.environ.get("DB_URL", "sqlite:///storefront.sqlite")
    return url.replace("postgres://", "postgresql+psycopg://", 1)


def redis_url() -> str:
    # A Valkey URL added before Oct 30, 2026 starts with valkey://, which redis-py doesn't accept.
    url = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
    return url.replace("valkey://", "redis://", 1)


def create_app() -> Flask:
    app = Flask(__name__)
    app.config["SQLALCHEMY_ENGINES"] = {"default": database_url()}
    app.config["CACHE_TYPE"] = "RedisCache"
    app.config["CACHE_REDIS_URL"] = redis_url()
    app.config["CACHE_DEFAULT_TIMEOUT"] = 300
    # Loads every FLASK_* environment variable, for example FLASK_SECRET_KEY.
    app.config.from_prefixed_env()

    db.init_app(app)
    alembic.init_app(app)
    cache.init_app(app)

    @app.get("/items")
    def list_items():
        items = db.session.scalars(sa.select(Item).order_by(Item.id))
        return [{"id": item.id, "name": item.name} for item in items]

    @app.post("/items")
    def create_item():
        item = Item(name=request.json["name"])
        db.session.add(item)
        db.session.commit()
        return {"id": item.id, "name": item.name}, 201

    @app.get("/cached")
    @cache.cached(timeout=120)
    def cached():
        return f"Generated at {datetime.now(timezone.utc):%H:%M:%S}"

    @app.get("/account")
    def account():
        response = make_response("Your account")
        response.cache_control.no_store = True
        return response

    @app.get("/catalog")
    def catalog():
        response = make_response("Catalog")
        response.cache_control.public = True
        response.cache_control.max_age = 600
        return response

    return app