[本記事の内容]

環境構築（Mac)：Homebrew で Redis をインストールし、仮想環境を用いて Python プロジェクトを作成。
実装例：Flask と SQLite を組み合わせたシンプルな商品情報 API でキャッシュ機能を実装。
Redis CLI 操作：redis-cli を使って接続確認、キーの確認、TTL の変更などを実施。
デモ操作：API エンドポイントにアクセスしてキャッシュの動作を確認する手順を紹介。

デモイメージ
初回アクセス時のキャッシュなしの流れと、再度アクセス時にキャッシュが利用される流れを示しています。

※ Redis とは

高速アクセス
Redis はメモリ上にデータを保持するため、ディスクベースのデータベースよりも高速にデータを読み書きできます。
データ構造
キーとバリューのシンプルな形式でデータを管理し、文字列、リスト、ハッシュ、セット、ソート済みセットなどのデータ構造をサポートします。
TTL (Time To Live)
キャッシュされたデータは、有効期限を設定して自動的に削除することができます。

1. 環境構築（Mac)

1-1. Redis のインストールと起動

Redis をインストールします。
```
brew update
brew install redis
```
Redis をバックグラウンドサービスとして起動するには、以下を実行します。
```
brew services start redis
```
または、手動で起動する場合は別のターミナルで以下を実行してください。
```
redis-server
```

1-2. Python 仮想環境のセットアップ

プロジェクト用のディレクトリを作成して移動します。
```
mkdir shop_api_project
cd shop_api_project
```

Python の仮想環境を作成して有効化します。

python3 -m venv venv
source venv/bin/activate   # Windows の場合は venv\Scripts\activate

必要なパッケージをインストールします。
```
pip install Flask redis Flask-SQLAlchemy
```
下記のソースコードを app.py として保存します（後述のコード例をコピー＆ペーストしてください）。
API サーバーを起動します。
```
python app.py
```

1-4. Redis CLI を使った確認

Redis CLI に接続するには、別のターミナルで以下を実行します。
```
redis-cli
```
接続後、ping や keys *、get <キー名> などのコマンドで中身を確認できます。
```
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> keys *
1) "product:1"
```

2. Flask と Redis を使ったキャッシュ実装例

以下は、SQLite を使った商品情報管理と、Redis によるキャッシュ機能を組み合わせた Flask API の例です。
この例では、DB に存在する商品情報を Redis にキャッシュして、API のレスポンスを高速化します。

from flask import Flask, jsonify
import redis
import time
import logging
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# ログ設定：DEBUGレベルで詳細なログを出力
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logging.debug("アプリケーション開始")

# Redis サーバーへの接続設定
try:
    cache = redis.Redis(host='localhost', port=6379, db=0)
    cache.ping()  # 接続確認
    logging.debug("Redis に正常に接続しました")
except Exception as e:
    logging.error("Redis への接続に失敗しました: %s", e)
    raise

# SQLAlchemy（SQLite）の設定
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///shop.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

# 商品情報を管理するテーブルモデルの定義
class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    description = db.Column(db.String(255), nullable=True)
    price = db.Column(db.Float, nullable=False)

# データベースとテーブルを作成（存在しなければ）
with app.app_context():
    db.create_all()
    # 初回起動時にサンプル商品を挿入
    if Product.query.count() == 0:
        sample_product = Product(
            name="ワイヤレスマウス",
            description="高精度のワイヤレスマウス。快適な操作性が特徴です。",
            price=2999.0
        )
        db.session.add(sample_product)
        db.session.commit()
        logging.debug("サンプル商品データを挿入しました")

def fetch_product_from_db(product_id):
    logging.debug("DB から商品 (id=%s) を取得開始", product_id)
    # 負荷のある処理のシミュレーション（5秒待機）
    time.sleep(5)
    product = Product.query.get(product_id)
    if product:
        result = {
            "id": product.id,
            "name": product.name,
            "description": product.description,
            "price": product.price
        }
        logging.debug("DB から取得した商品情報: %s", result)
        return result
    else:
        logging.debug("商品 (id=%s) は DB に存在しません", product_id)
        return None

# キャッシュの有効期限（秒）を変数で管理
CACHE_TTL = 60

@app.route('/api/products/<int:product_id>')
def get_product(product_id):
    cache_key = f"product:{product_id}"
    logging.debug("キャッシュチェックを実行中。キー: %s", cache_key)
    cached_data = cache.get(cache_key)
    if cached_data:
        product_data = cached_data.decode('utf-8')
        logging.debug("キャッシュヒット。取得したデータ: %s", product_data)
        return jsonify({
            "source": "cache",
            "data": product_data
        })

    logging.debug("キャッシュミス。DB から商品情報を取得します")
    product = fetch_product_from_db(product_id)
    if product:
        # キャッシュにはシリアライズした文字列を保存（ここでは str() で簡易的に保存）
        product_str = str(product)
        cache.setex(cache_key, CACHE_TTL, product_str)
        ttl = cache.ttl(cache_key)
        logging.debug("DB から取得した商品情報をキャッシュに保存しました。キー: %s, TTL: %s秒", cache_key, ttl)
        return jsonify({
            "source": "db",
            "data": product
        })
    else:
        return jsonify({
            "source": "db",
            "error": "商品が見つかりません"
        }), 404

if __name__ == '__main__':
    logging.debug("Flask サーバーを起動します")
    app.run(debug=True)

3. Redis CLI を使った基本操作

Redis CLI を使うと、ターミナル上で Redis に保存されているデータの確認や操作が可能です。

接続と基本操作の例

Redis CLI に接続
```
redis-cli
```
接続確認
```
127.0.0.1:6379> ping
PONG
```
キーの一覧表示
```
127.0.0.1:6379> keys *
1) "product:1"
```

特定のキーの値を取得

127.0.0.1:6379> get "product:1"
"商品情報が保存された文字列"

TTL の確認と変更

現在の TTL を確認するには:

127.0.0.1:6379> ttl product:1
(integer) 55

有効期限を変更するには:

127.0.0.1:6379> expire product:1 120
(integer) 1

4. API デモの操作手順

Flask サーバーの起動
既に python app.py を実行して、Flask サーバーが http://127.0.0.1:5000 で起動している状態にしてください。
ブラウザまたは API クライアントでアクセス
ブラウザを開き、以下の URL にアクセスします。
http://127.0.0.1:5000/api/products/1
初回アクセスの動作
- 初回のアクセスでは、キャッシュが存在しないため、DB から商品情報が取得され、キャッシュに保存されます。
- レスポンスの JSON には "source": "db" と表示され、商品情報が返されます。
再度アクセスしてキャッシュ確認
同じ URL に再度アクセスすると、キャッシュに保存されているため、
- レスポンスの JSON に "source": "cache" と表示され、即時にキャッシュされた商品情報が返されます。
Redis CLI での確認
別のターミナルで redis-cli を起動し、keys * や get "product:1" を使って、実際に Redis にキャッシュされたデータの内容や TTL を確認できます。

参考

https://qiita.com/yoh-nak/items/69c26e7d627e2b5a95e2
https://qiita.com/wind-up-bird/items/f2d41d08e86789322c71
https://qiita.com/takeh/items/8132977d9ed6f82a86f4
https://qiita.com/hirotakasasaki/items/9819a4e6e1f33f99213c