Valkey: Uso con Python

Cómo conectar Valkey con Python

👋 ¡Bienvenido a la documentación de Stackhero!

Stackhero ofrece una solución Valkey cloud lista para usar que proporciona una serie de beneficios, incluyendo:

• Interfaz web UI Redis Commander incluida.
• Tamaño y transferencias de mensajes ilimitados.
• Actualizaciones sin esfuerzo con solo un clic.
• Rendimiento óptimo y seguridad robusta gracias a una VM privada y dedicada.

Ahorra tiempo y simplifica tu vida: ¡solo toma 5 minutos probar la solución de Valkey cloud hosting de Stackhero!

Elegir una biblioteca de Python para la integración de Valkey

Para integrar su aplicación con Valkey de manera fluida, considere usar la biblioteca redis para Python. Esta biblioteca es adecuada para conectarse a Valkey, un fork de código abierto de Redis. Puede instalarla fácilmente ejecutando:

pip install redis
pip freeze > requirements.txt

Conectar Python a Valkey

Exploremos cómo puede conectar su aplicación Python a Valkey. Aquí hay un ejemplo sencillo usando la configuración predeterminada:

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
)

Para una mayor seguridad, se recomienda gestionar las credenciales utilizando variables de entorno. Aquí le mostramos cómo hacerlo:

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
)

Asegúrese de que sus variables de entorno incluyan una entrada como esta: STACKHERO_VALKEY_URL_TLS=rediss://default:<password>@<XXXXXX>.stackhero-network.com:<PORT_TLS>.

Usar Pub/Sub con Valkey y Python

Aprovechar la funcionalidad Publish/Subscribe (Pub/Sub) en Python con Valkey es sencillo. Aquí hay un ejemplo:

import redis

# Conectar a 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
)

# Crear una instancia de PubSub
p = r.pubsub()

# Suscribirse al canal "test"
p.subscribe('test')

# Publicar un mensaje en el canal "test"
r.publish('test', 'Este es un mensaje de prueba')

# Obtener el primer mensaje disponible del canal "test"
p.get_message()

# Cancelar la suscripción del canal "test"
p.unsubscribe('test')

Ejemplos avanzados de Valkey Pub/Sub con Python

Amplíe sus capacidades de Pub/Sub con estos ejemplos avanzados:

# Crear una instancia de PubSub e ignorar los mensajes de suscripción
p = r.pubsub(ignore_subscribe_messages=True)

# Suscribirse a múltiples canales
p.subscribe('test-1', 'test-2', ...)

# Cancelar la suscripción de múltiples canales
p.unsubscribe('test-1', 'test-2', ...)

# También puede usar "unsubscribe" sin argumentos para desconectarse de todos los canales suscritos
p.unsubscribe()

# Suscribirse a canales usando un patrón
p.psubscribe('my-*')

# Cancelar la suscripción de canales usando un patrón
p.punsubscribe('my-*')

Evitar errores "Connection Closed by Server"

El error redis.exceptions.ConnectionError: Connection closed by server puede ocurrir si su aplicación no interactúa con Valkey durante un período, lo que lleva a una desconexión automática. Para evitar esto, puede establecer el parámetro health_check_interval como se muestra a continuación:

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
)

Al usar la función Pub/Sub de Valkey, la biblioteca redis-py espera que funciones como get_message() o listen() se llamen con más frecuencia que el health_check_interval. En nuestro ejemplo, hemos establecido este intervalo en 10 segundos, así que asegúrese de llamar a get_message() o listen() al menos una vez cada 10 segundos (consulte la documentación oficial de redis-py).

Si esto no se hace, podría encontrar el mismo error de conexión. Para evitar esto, considere llamar regularmente a check_health().

Aquí le mostramos cómo puede implementarlo:

import redis
import threading

# Conectar a 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
)

# Crear una instancia de PubSub
p = r.pubsub()

# Suscribirse al canal "test"
p.subscribe('test')

# Crear una función que llame a `check_health` cada 5 segundos
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# Llamar a la función valkey_auto_check
valkey_auto_check(p)