Valkey: Utilisation avec Python

Comment connecter Valkey avec Python

👋 Bienvenue sur la documentation de Stackhero !

Stackhero propose une solution Valkey cloud prête à l'emploi qui offre de nombreux avantages, notamment :

Interface web Redis Commander incluse.

Taille et transferts de messages illimités.

Mises à jour simplifiées en un clic.

Performance optimale et sécurité renforcée grâce à une VM privée et dédiée.

Gagnez du temps et simplifiez-vous la vie : il suffit de 5 minutes pour essayer la solution Valkey cloud hosting de Stackhero !

Choisir une bibliothèque Python pour l'intégration de Valkey

Pour intégrer facilement votre application avec Valkey, il est recommandé d'utiliser la bibliothèque redis pour Python. Cette bibliothèque est parfaitement adaptée pour se connecter à Valkey, un fork open source de Redis. Vous pouvez l'installer simplement en exécutant :

pip install redis
pip freeze > requirements.txt

Connecter Python à Valkey

Voyons comment connecter votre application Python à Valkey. Voici un exemple simple utilisant les paramètres par défaut :

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
)

Pour renforcer la sécurité, il est conseillé de gérer les identifiants via des variables d'environnement. Voici comment procéder :

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
)

Assurez-vous que vos variables d'environnement contiennent une entrée comme celle-ci : STACKHERO_VALKEY_URL_TLS=rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>.

Utiliser Pub/Sub avec Valkey et Python

Exploiter la fonctionnalité Publish/Subscribe (Pub/Sub) avec Python et Valkey est très simple. Voici un exemple :

import redis

# Connexion à 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
)

# Créer une instance PubSub
p = r.pubsub()

# S'abonner au canal "test"
p.subscribe('test')

# Publier un message sur le canal "test"
r.publish('test', 'This is a test message')

# Récupérer le premier message disponible du canal "test"
p.get_message()

# Se désabonner du canal "test"
p.unsubscribe('test')

Exemples avancés de Pub/Sub Valkey avec Python

Développez vos usages Pub/Sub avec ces exemples avancés :

# Créer une instance PubSub et ignorer les messages d'abonnement
p = r.pubsub(ignore_subscribe_messages=True)

# S'abonner à plusieurs canaux
p.subscribe('test-1', 'test-2', ...)

# Se désabonner de plusieurs canaux
p.unsubscribe('test-1', 'test-2', ...)

# Vous pouvez aussi utiliser "unsubscribe" sans argument pour se désabonner de tous les canaux
p.unsubscribe()

# S'abonner à des canaux via un pattern
p.psubscribe('my-*')

# Se désabonner de canaux via un pattern
p.punsubscribe('my-*')

Eviter les erreurs "Connection Closed by Server"

L'erreur redis.exceptions.ConnectionError: Connection closed by server peut survenir si votre application n'interagit pas avec Valkey pendant un certain temps, ce qui entraîne une déconnexion automatique. Pour éviter cela, vous pouvez définir le paramètre health_check_interval comme ci-dessous :

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
)

Lorsque vous utilisez la fonctionnalité Pub/Sub de Valkey, la bibliothèque redis-py attend que des fonctions comme get_message() ou listen() soient appelées plus fréquemment que la valeur de health_check_interval. Dans notre exemple, cet intervalle est fixé à 10 secondes, il est donc important d'appeler get_message() ou listen() au moins une fois toutes les 10 secondes (voir la documentation officielle de redis-py).

Si cela n'est pas fait, vous risquez de rencontrer la même erreur de connexion. Pour l'éviter, pensez à appeler régulièrement check_health().

Voici comment l'implémenter :

import redis
import threading

# Connexion à 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
)

# Créer une instance PubSub
p = r.pubsub()

# S'abonner au canal "test"
p.subscribe('test')

# Créer une fonction qui appelle `check_health` toutes les 5 secondes
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# Appeler la fonction valkey_auto_check
valkey_auto_check(p)