本記事では、インメモリデータベースとして人気の高いRedisを、Pythonから使う方法を初心者向けにやさしく解説します。
基本的な考え方から導入手順、主要なデータ型の扱い方、そして実際にPythonコードを書く練習問題まで盛り込みました。この記事を通じて、以下のスキルが身につきます。
- Redisの概要と特徴を理解する
- PythonでRedisサーバーに接続し、基礎的な操作を行う
- 各種データ型の使い分けができる
- TTL(有効期限)の設定方法を学ぶ
- 簡単な演習問題を通して理解を定着させる
Redisとは何か
Redis(Remote Dictionary Server)は、メモリ上で動作する高速なキー・バリュー型データストアです。主に以下の用途で利用されます。
- キャッシュ(例:Webページのキャッシュ/APIレスポンスのキャッシュ)
- セッション管理
- リアルタイムランキング(Sorted Set)
- メッセージキュー(Pub/Sub 機能)
Redisはシングルスレッドで動作しますが、イベントループと効率的なデータ構造により、非常に高いパフォーマンスを実現しています。
環境構築
Redis サーバーのインストール
macOS(Homebrewの場合)
brew update
brew install redis
brew services start redis
Ubuntu/Debian
sudo apt update
sudo apt install redis-server
sudo systemctl enable redis-server
sudo systemctl start redis-server
インストール後、redis-cli コマンドを実行して、Redisサーバーに接続できるか確認してみましょう。
$ redis-cli ping
PONG
Python用クライアントライブラリ(redis-py)のインストール
PythonからRedisに接続するためのライブラリとして、公式サポートのある redis-py を使います。以下のコマンドでインストールします。
pip install redis
Redisの基本操作
キーと値の登録・取得
Redisはシンプルなキー・バリュー型データベースです。文字列型(String)の基本操作は以下の通りです。
- SET: キーに値をセット
- GET: キーから値を取得
> SET mykey "こんにちは"
OK
> GET mykey
"こんにちは"
データ型
Redisには主要なデータ型が5つあります。用途に合わせて使い分けましょう。
- String
- 単純な文字列や数字を保持
- List
- 挿入順序を保持するリスト
- LPUSH, RPUSH, LPOP, RPOP 等
- Set
- 重複を許さない集合
- SADD, SMEMBERS, SREM 等
- Hash
- フィールドと値のマッピング(辞書)
- HSET, HGETALL, HDEL 等
- Sorted Set
- スコア付き集合で、自動的にソートされる
- ZADD, ZRANGE, ZREM 等
有効期限(TTL)の設定
Redisではキーに対して有効期限(TTL: Time To Live)を設定できます。キャッシュ用途などに便利です。
> SET tempkey "一時データ"
OK
> EXPIRE tempkey 60 # 60秒後に自動削除
(integer) 1
> TTL tempkey
(integer) 57
PythonからRedisを使ってみよう
以下ではPythonスクリプト内での接続と操作例を示します。
import redis
# Redisサーバーに接続(デフォルトhost=localhost, port=6379)
r = redis.Redis()
# String型の操作
r.set('greeting', 'こんにちは Redis')
print(r.get('greeting').decode()) # => こんにちは Redis
# List型の操作
r.rpush('tasks', 'task1', 'task2', 'task3')
tasks = r.lrange('tasks', 0, -1)
print([t.decode() for t in tasks]) # => ['task1', 'task2', 'task3']
# Hash型の操作
r.hset('user:100', mapping={'name': 'Alice', 'age': '30'})
user = r.hgetall('user:100')
print({k.decode(): v.decode() for k, v in user.items()})
# => {'name': 'Alice', 'age': '30'}
# TTLの設定
r.set('cache_item', 'value', ex=120) # exパラメータで秒数指定
print(r.ttl('cache_item'))
エラー対策とベストプラクティス
接続エラー対策
redis.exceptions.ConnectionError のハンドリング
データ型間違いの防止
キーの命名規則: type:identifier 形式で分類する
- 例: user:100, session:abcd1234
パイプラインの活用
複数コマンドをまとめて送信し、往復回数を減らす。
with r.pipeline() as pipe:
pipe.set('a', 1)
pipe.set('b', 2)
pipe.get('a')
pipe.get('b')
results = pipe.execute()
トランザクション
WATCH + MULTI + EXEC を使った楽観的ロック
演習問題
以下の内容をRedisとPythonを使って実装してみましょう。
- ユーザー投稿のカウント
Hash型を使って、ユーザーIDごとの投稿数を管理する関数 increment_post_count(user_id) を作成してください。 - ランキング機能
Sorted Setを使って、ユーザーのスコアランキングを実装する関数 add_score(user_id, score) と、上位5ユーザーを取得する関数 get_top5() を作成してください。 - 一時ログ
List型を使って、最新10件の操作ログを保持する関数 push_log(message) を実装してください。古いログは自動で削除されるようにします。 - キャッシュ切れの確認
String型でキャッシュを保存し、キーが切れた場合に再生成するラッパー関数 get_or_set_cache(key, generator_func, ex_seconds) を作成してください。
演習問題 解答例
import redis
r = redis.Redis()
# 1. ユーザー投稿のカウント
def increment_post_count(user_id):
key = f"user:{user_id}:posts"
return r.hincrby("user_posts", key, 1)
# 2. ランキング機能
def add_score(user_id, score):
# Sorted Set 'user_scores' にスコアを追加
r.zincrby("user_scores", score, user_id)
def get_top5():
# スコアが高い順に上位5件を取得 (reverse=True)
top = r.zrevrange("user_scores", 0, 4, withscores=True)
return [(user.decode(), int(score)) for user, score in top]
# 3. 一時ログ
def push_log(message):
key = "operation_logs"
# 最新10件のみ保持
r.lpush(key, message)
r.ltrim(key, 0, 9)
# 4. キャッシュ切れの確認
def get_or_set_cache(key, generator_func, ex_seconds=60):
val = r.get(key)
if val is not None:
return val.decode()
# キャッシュがない場合は generator_func() で取得しセット
new_val = generator_func()
r.set(key, new_val, ex=ex_seconds)
return new_val
まとめ
本記事では、Redisの概要からインストール、主要なデータ型、Pythonからの利用方法、演習問題までを解説しました。
Redisは高速かつ使い勝手の良いインメモリデータストアとして、多くのWebサービスやアプリケーションで採用されています。
ぜひ、実際に手を動かしながらRedisの操作に慣れてみてください。応用的な機能(Pub/Sub、ストリーム、トランザクションなど)にもチャレンジすると、さらに活用の幅が広がります。