PythonでRedisを操作する方法｜redis-pyの接続・データ型・TTL・パイプライン・キャッシュの実装

Python から Redis を操作するには、公式のクライアントライブラリ redis-py（パッケージ名 redis）を使い、redis.Redis で接続して set()・get()・hset() などデータ型ごとのメソッドを呼び出します。この記事では、安全な接続設定から、文字列・ハッシュ・リスト・セット・ソート済みセットの基本操作、有効期限（TTL）、パイプライン、キャッシュとロックの実装、本番運用でつまずきやすい点までを説明します。

準備：redis-py のインストールと接続

pip install redis
python -c "import redis; print(redis.__version__)"

接続先の Redis はローカルか、同じネットワーク内のサーバーを想定します。この記事のコードは redis-py 5 系以降で動く書き方にしています（2026年10月現在）。なお、Redis は 2024年にライセンスが変更され、その後のバージョンで再びオープンソースのライセンスが選べるようになったほか、互換のフォーク（Valkey）も広く使われています。redis-py は多くの場合 Valkey にもそのまま接続できますが、利用するサーバーのバージョンと、使いたいコマンドが対応しているかは事前に確認してください。

接続の基本形です。

import os
import redis

r = redis.Redis(
    host=os.getenv("REDIS_HOST", "127.0.0.1"),
    port=int(os.getenv("REDIS_PORT", "6379")),
    db=int(os.getenv("REDIS_DB", "0")),
    password=os.getenv("REDIS_PASSWORD") or None,
    decode_responses=True,       # bytes ではなく str で受け取る
    socket_connect_timeout=3,
    socket_timeout=3,
)

print(r.ping())   # True が返れば接続できている

設定のポイントは次のとおりです。

• decode_responses=True：指定しないと値が b"value" のような bytes で返ります。テキストを扱うだけなら True にしておくと、毎回 decode する手間が省けます。画像などのバイナリを保存する場合は、そのクライアントだけ False にします。
• タイムアウト：socket_connect_timeout と socket_timeout を指定しないと、Redis が応答しないときに処理が止まったままになることがあります。
• パスワードはコードに書かない：環境変数などから読みます。Redis 6 以降の ACL でユーザーを分けている場合は username= も指定します。

接続先を URL でまとめて指定したい場合は redis.from_url("redis://127.0.0.1:6379/0", decode_responses=True) が使えます。TLS で接続する場合は rediss:// で始まる URL を使います。

redis.Redis のインスタンスは内部に接続プールを持っています。リクエストのたびに作り直すのではなく、アプリ全体で1つを使い回すのが基本です。

データ型ごとの基本操作

Redis はキーに対して値を持つ仕組みですが、値にはいくつかの型があり、型ごとにコマンドが分かれています。用途に合った型を選ぶと、処理が単純になります。

# 文字列：単純な値やカウンター
r.set("site:status", "ok")
r.get("site:status")                  # 'ok'
r.incr("counter:pageview")            # 1, 2, 3 ... と増える

# ハッシュ：項目を持つオブジェクト
r.hset("user:1001", mapping={"name": "Sato", "plan": "basic"})
r.hget("user:1001", "plan")           # 'basic'
r.hgetall("user:1001")                # {'name': 'Sato', 'plan': 'basic'}

# リスト：順番のあるデータ（キューなど）
r.lpush("queue:mail", "job-1", "job-2")
r.rpop("queue:mail")                  # 'job-1'（先に入れたものから取り出す）

# セット：重複のない集合
r.sadd("tags:post:10", "php", "wordpress", "php")
r.smembers("tags:post:10")            # {'php', 'wordpress'}

# ソート済みセット：スコア付きのランキング
r.zadd("ranking:daily", {"post:10": 120, "post:11": 85})
r.zincrby("ranking:daily", 5, "post:11")
r.zrange("ranking:daily", 0, 9, desc=True, withscores=True)
# [('post:10', 120.0), ('post:11', 90.0)]

キー名は 種類:ID のようにコロンで区切るのが慣習です。後から「user で始まるキーだけ調べたい」といった場面で探しやすくなります。

ハッシュに保存した値はすべて文字列として返ります。数値として使うときは int() などで変換します。入れ子の辞書やリストを保存したい場合は、json.dumps() で文字列にしてから文字列型に入れるのが簡単です。

有効期限（TTL）を付ける

キャッシュや一時的なデータには、有効期限を付けて自動で消えるようにします。付け忘れると、使われなくなったキーがメモリに残り続けます。

r.set("tmp:import-status:abc", "pending", ex=300)   # 300秒後に消える
r.ttl("tmp:import-status:abc")                      # 残り秒数（期限なしは -1、キーがなければ -2）

r.hset("user:1001:profile", mapping={"name": "Sato"})
r.expire("user:1001:profile", 3600)           # 既存のキーに期限を付ける

# まだ存在しないときだけ書き込む（重複実行の防止に使える）
r.set("job:daily-report:2026-10-28", "running", nx=True, ex=600)

set() の nx=True は「キーがないときだけ書き込む」指定です。戻り値が True なら書き込めた、None ならすでに存在していた、という意味になるので、同じ処理が二重に走るのを防ぐ目印として使えます。

パイプラインで往復の回数を減らす

Redis のコマンドは1回ごとにネットワークの往復が発生します。大量のコマンドを続けて送るときは、パイプラインでまとめて送ると速くなります。

with r.pipeline() as pipe:
    for post_id in range(1, 101):
        pipe.hset(f"post:{post_id}", mapping={"views": 0})
        pipe.expire(f"post:{post_id}", 86400)
    results = pipe.execute()

pipeline() は既定で transaction=True になっており、まとめたコマンドが MULTI/EXEC で囲まれて一連の処理として実行されます。速さだけが目的で一括性が不要なら pipeline(transaction=False) も使えます。ただし、これは途中のコマンドが失敗したときに前のコマンドを取り消す仕組みではない点に注意してください。

キャッシュとロックを実装する

よく使うのが「キャッシュにあればそれを返し、なければ元のデータを取得して保存する」パターンです。

import json


def get_post_summary(post_id: int) -> dict:
    key = f"cache:post-summary:{post_id}"
    cached = r.get(key)
    if cached is not None:
        return json.loads(cached)

    data = load_post_summary_from_db(post_id)   # 重い処理（自分の関数に置き換える）
    r.set(key, json.dumps(data, ensure_ascii=False), ex=600)
    return data

元データを更新したときは r.delete(key) でキャッシュを消すと、次のアクセスで新しい内容が保存されます。

複数のプロセスで同じ処理を同時に実行させたくない場合は、redis-py の lock() が使えます。

from redis.exceptions import LockError

try:
    with r.lock("lock:rebuild-sitemap", timeout=60, blocking_timeout=5):
        rebuild_sitemap()                        # 同時に1つだけ実行したい処理
except LockError:
    print("ほかのプロセスが実行中のためスキップ")

timeout はロックが自動で外れるまでの秒数です。処理中にプロセスが落ちてもロックが残り続けないようにするための値なので、処理にかかる時間より長めに設定します。

本番で気を付けること

KEYS コマンドを使わない：r.keys("user:*") はすべてのキーを一度に調べるため、キーが多いと Redis 全体の処理が止まります。代わりに少しずつ調べる scan_iter() を使います。

for key in r.scan_iter(match="cache:post-summary:*", count=500):
    r.delete(key)

共有している Redis で FLUSHDB を実行しない：WordPress のオブジェクトキャッシュ（Redis Object Cache プラグインなど）と同じ Redis を使っている場合、flushdb() や flushall() を実行するとサイトのキャッシュまで消えます。Python のスクリプト用には別のデータベース番号を使うか、キー名に専用の接頭辞を付け、消すときは接頭辞で絞って scan_iter() で削除します。

接続エラーを想定する：Redis が再起動している間などは redis.exceptions.ConnectionError や TimeoutError が発生します。キャッシュとして使っているなら、Redis に接続できないときは元データを直接返すようにしておくと、Redis の停止がサービス全体の停止につながりません。

外部に公開しない：Redis は 127.0.0.1 や内部ネットワークだけで待ち受け、ファイアウォールで 6379 番ポートを外部から閉じます。パスワードか ACL を設定し、別のサーバーから接続する場合は TLS を使います。

非同期処理では redis.asyncio を使う：asyncio を使うアプリ（FastAPI の async 関数など）では、import redis.asyncio as redis で同じ形の非同期クライアントが使えます。メソッドは await r.get("key") のように呼び出します。

まとめ

• redis-py の redis.Redis で接続し、decode_responses=True とタイムアウトを指定します。インスタンスは使い回します。
• 文字列・ハッシュ・リスト・セット・ソート済みセットを用途に合わせて使い分け、一時データには ex や expire() で期限を付けます。
• 大量のコマンドはパイプラインでまとめ、同時実行の防止には lock() を使います。
• KEYS と、共有している Redis での FLUSHDB は避け、scan_iter() と接頭辞で対象を絞ります。
• Redis は外部に公開せず、パスワードや ACL、必要に応じて TLS を設定します。

• Redisのデータ型と基本コマンド
• WordPressのオブジェクトキャッシュをRedisで高速化する方法