Valkey: Avoiding "Connection Closed by Server" errors

This documentation is part of the Using with Python guide. You can view the complete guide here: How to connect Valkey with Python.

👋 Welcome to the Stackhero documentation!

Stackhero offers a ready-to-use Valkey cloud solution that provides a host of benefits, including:

• Redis Commander web UI included.
• Unlimited message size and transfers.
• Effortless updates with just a click.
• Optimal performance and robust security powered by a private and dedicated VM.

Save time and simplify your life: it only takes 5 minutes to try Stackhero's Valkey cloud hosting solution!

The error redis.exceptions.ConnectionError: Connection closed by server may occur if your app does not interact with Valkey for a period, leading to an automatic disconnection. To prevent this, you can set the health_check_interval parameter as shown below:

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
)

When using Valkey's Pub/Sub feature, the redis-py library expects functions like get_message() or listen() to be called more frequently than the health_check_interval. In our example, we have set this interval to 10 seconds, so make sure to call get_message() or listen() at least once every 10 seconds (refer to the redis-py official documentation).

If this is not done, you might encounter the same connection error. To avoid this, consider regularly calling check_health().

Here is how you can implement it:

import redis
import threading

# Connect to 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
)

# Create a PubSub instance
p = r.pubsub()

# Subscribe to the channel "test"
p.subscribe('test')

# Create a function that will call `check_health` every 5 seconds
def valkey_auto_check(p):
  t = threading.Timer(5, valkey_auto_check, [p])
  t.start()
  p.check_health()

# Call the valkey_auto_check function
valkey_auto_check(p)