Valkey: Pythonでの使用

ValkeyをPythonで接続する方法

👋 Stackheroのドキュメントへようこそ！

Stackheroは、数多くの利点を提供するValkeyクラウドソリューションをすぐに利用できる形で提供しています。

Redis Commander Web UIが含まれています。

メッセージサイズと転送が無制限。

ワンクリックで簡単にアップデート。

プライベートで専用のVMによる最適なパフォーマンスと強力なセキュリティ。

時間を節約し、生活を簡素化：StackheroのValkeyクラウドホスティングソリューションを試すのに5分しかかかりません！

Valkey統合のためのPythonライブラリの選択

アプリケーションをValkeyとシームレスに統合するには、Python用のredisライブラリを使用することを検討してください。このライブラリは、RedisのオープンソースフォークであるValkeyへの接続に適しています。以下のコマンドで簡単にインストールできます。

pip install redis
pip freeze > requirements.txt

PythonをValkeyに接続する

PythonアプリケーションをValkeyに接続する方法を見てみましょう。デフォルト設定を使用した簡単な例を以下に示します。

import redis

r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

セキュリティを強化するために、環境変数を使用して資格情報を管理することをお勧めします。以下にその方法を示します。

import os
import redis

r = redis.from_url(
  os.environ.get("STACKHERO_VALKEY_URL_TLS"),
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

環境変数に次のようなエントリが含まれていることを確認してください: STACKHERO_VALKEY_URL_TLS=rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>.

ValkeyとPythonでのPub/Subの使用

PythonでValkeyのPublish/Subscribe (Pub/Sub)機能を活用するのは簡単です。以下に例を示します。

import redis

# Valkeyに接続
r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  socket_keepalive=True,
  retry_on_timeout=True
)

# PubSubインスタンスを作成
p = r.pubsub()

# チャンネル"test"にサブスクライブ
p.subscribe('test')

# チャンネル"test"にメッセージを公開
r.publish('test', 'This is a test message')

# チャンネル"test"から最初の利用可能なメッセージを取得
p.get_message()

# チャンネル"test"のサブスクライブを解除
p.unsubscribe('test')

Valkey Pub/Subの高度な例

これらの高度な例でPub/Subの機能を拡張してください。

# PubSubインスタンスを作成し、サブスクライブメッセージを無視
p = r.pubsub(ignore_subscribe_messages=True)

# 複数のチャンネルにサブスクライブ
p.subscribe('test-1', 'test-2', ...)

# 複数のチャンネルのサブスクライブを解除
p.unsubscribe('test-1', 'test-2', ...)

# 引数なしで"unsubscribe"を使用して、すべてのサブスクライブされたチャンネルから切断することもできます
p.unsubscribe()

# パターンを使用してチャンネルにサブスクライブ
p.psubscribe('my-*')

# パターンを使用してチャンネルのサブスクライブを解除
p.punsubscribe('my-*')

"Connection Closed by Server"エラーの回避

アプリが一定期間Valkeyとやり取りしない場合、redis.exceptions.ConnectionError: Connection closed by serverエラーが発生することがあります。これを防ぐために、以下のようにhealth_check_intervalパラメータを設定できます。

r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

ValkeyのPub/Sub機能を使用する際、redis-pyライブラリはget_message()やlisten()のような関数がhealth_check_intervalより頻繁に呼び出されることを期待しています。この例では、この間隔を10秒に設定しているので、少なくとも10秒に1回はget_message()またはlisten()を呼び出してください（redis-py公式ドキュメントを参照）。

これを行わないと、同じ接続エラーが発生する可能性があります。これを避けるために、定期的にcheck_health()を呼び出すことを検討してください。

以下にその実装方法を示します。

import redis
import threading

# Valkeyに接続
r = redis.from_url(
  'rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>',
  health_check_interval=10,
  socket_connect_timeout=5,
  retry_on_timeout=True,
  socket_keepalive=True
)

# PubSubインスタンスを作成
p = r.pubsub()

# チャンネル"test"にサブスクライブ
p.subscribe('test')

# 5秒ごとに`check_health`を呼び出す関数を作成
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# valkey_auto_check関数を呼び出す
valkey_auto_check(p)