Connect your Python application to a Redis database

redis-py guide (Python)
https://redis.io/docs/latest/develop/clients/redis-py/

redis-py (Redis Python client)
https://github.com/redis/redis-py

The Python interface to the Redis key-value store.

redis-py is the Python client for Redis. redis-py requires a running Redis server.

The sections below explain how to install redis-py and connect your Python application to a Redis database.

1. Installation

To install redis-py, enter:

pip install redis

(base) yongqiang@yongqiang:~$ pip install redis
Collecting redisDownloading redis-6.4.0-py3-none-any.whl (279 kB)━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 279.8/279.8 kB 316.2 kB/s eta 0:00:00
Installing collected packages: redis
Successfully installed redis-6.4.0
(base) yongqiang@yongqiang:~$

For faster performance, install Redis with hiredis support. This provides a compiled response parser, and for most cases requires zero code changes. By default, if hiredis >= 1.0 is available, redis-py attempts to use it for response parsing.

The Python distutils packaging scheme is no longer part of Python 3.12 and greater. If you’re having difficulties getting redis-py installed in a Python 3.12 environment, consider updating to a recent release of redis-py.

pip install redis[hiredis]

2. Connect and test

Connect to localhost on port 6379, set a value in Redis, and retrieve it.

All responses are returned as bytes in Python. To receive decoded strings, set decode_responses=True.
redis 取出的結果默認是字節，可以設定 decode_responses=True 改成字符串。

#!/usr/bin/env python
# coding=utf-8import redisr = redis.Redis(host='localhost', port=6379, decode_responses=True)
r.set('name', 'yongqiang')print(f"name: {r['name']}")print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")

redis 提供兩個類 Redis 和 StrictRedis, StrictRedis 用于實現大部分官方的命令，Redis 是 StrictRedis 的子類，用于向后兼用舊版本。

#!/usr/bin/env python
# coding=utf-8import redisr = redis.StrictRedis(host='localhost', port=6379, decode_responses=True)
r.set('name', 'yongqiang')print(f"name: {r['name']}")print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
name: yongqiang
type: <class 'str'>
name: yongqiangProcess finished with exit code 0

3. Connection Pools

By default, redis-py uses a connection pool to manage connections. Each instance of a Redis class receives its own connection pool.

redis-py 使用 connection pool 來管理對一個 redis-server 的所有連接，避免每次建立、釋放連接的開銷。默認情況下，每個 Redis 實例都會維護一個自己的連接池。可以直接建立一個連接池，然后作為 Redis 參數，這樣就可以實現多個 Redis 實例共享一個連接池。

#!/usr/bin/env python
# coding=utf-8import redispool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)r.set('name', 'yongqiang')print(f"name: {r['name']}")print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
name: yongqiang
type: <class 'str'>
name: yongqiangProcess finished with exit code 0

4. Redis Commands

https://redis.readthedocs.io/en/stable/commands.html

4.1. set(name, value, ex=None, px=None, nx=False, xx=False, keepttl=False, get=False, exat=None, pxat=None)

Set the value at key name to value.
在 Redis 中設置值，默認：不存在則創建，存在則修改。

https://redis.io/docs/latest/commands/set/

Parameters
name (Union[bytes, str, memoryview])
value (Union[bytes, bytearray, memoryview, str, int, float])
ex (Optional[Union[int, timedelta]]) - sets an expire flag on key name for ex seconds.
px (Optional[Union[int, timedelta]]) - sets an expire flag on key name for px milliseconds.
nx (bool) - if set to True, set the value at key name to value only if it does not exist. 如果設置為 True，則只有 name 不存在時，當前 set 操作才執行 (新建 / 添加)。
xx (bool) - if set to True, set the value at key name to value only if it already exists. 如果設置為 True，則只有 name 存在時，當前 set 操作才執行 (修改)。
keepttl (bool) - if True, retain the time to live associated with the key. (Available since Redis 6.0)
get (bool) - if True, set the value at key name to value and return the old value stored at key, or None if the key did not exist. (Available since Redis 6.2)
exat (Optional[Union[int, datetime]]) - sets an expire flag on key name for ex seconds, specified in unix time.
pxat (Optional[Union[int, datetime]]) - sets an expire flag on key name for ex milliseconds, specified in unix time.

Return type
Union[Awaitable[Any], Any]

• ex (Optional[Union[int, timedelta]]) - sets an expire flag on key name for ex seconds.

設置過期時間是 2 秒。2 秒后，鍵 name 的值就變成 None。

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)r.set('name', 'yongqiang', ex=2)print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")print("Start: %s" % time.ctime())
time.sleep(5)  # secs
print("End: %s" % time.ctime())print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
type: <class 'str'>
name: yongqiang
Start: Fri Aug 15 00:20:10 2025
End: Fri Aug 15 00:20:15 2025
name: NoneProcess finished with exit code 0

• px (Optional[Union[int, timedelta]]) - sets an expire flag on key name for px milliseconds.

設置過期時間是 200 milliseconds。200 milliseconds 后，鍵 name 的值就變成 None。

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)r.set('name', 'yongqiang', px=200)print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")print("Start: %s" % time.ctime())
time.sleep(2)  # secs
print("End: %s" % time.ctime())print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
type: <class 'str'>
name: yongqiang
Start: Fri Aug 15 00:24:02 2025
End: Fri Aug 15 00:24:04 2025
name: NoneProcess finished with exit code 0

• nx (bool) - if set to True, set the value at key name to value only if it does not exist.

如果設置為 True，則只有 name 不存在時，當前 set 操作才執行 (新建 / 添加)。

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)ret1 = r.set('name', 'yongqiang', nx=True)
print(f"ret1: {ret1}")ret2 = r.set('name', 'cheng', nx=True)
print(f"ret2: {ret2}")time.sleep(2)  # secs

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
ret1: True
ret2: NoneProcess finished with exit code 0

• xx (bool) - if set to True, set the value at key name to value only if it already exists.

如果設置為 True，則只有 name 存在時，當前 set 操作才執行 (修改)。

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)ret1 = r.set('name', 'yongqiang', xx=True)
print(f"ret1: {ret1}")ret2 = r.set('name', 'yongqiang')
print(f"ret2: {ret2}")ret3 = r.set('name', 'cheng', xx=True)
print(f"ret3: {ret3}")time.sleep(2)  # secs

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
ret1: None
ret2: True
ret3: TrueProcess finished with exit code 0

4.1.1. setnx(name, value)

Set the value of key name to value if key doesn’t exist.

https://redis.io/docs/latest/commands/setnx/

Parameters
name (Union[bytes, str, memoryview])
value (Union[bytes, bytearray, memoryview, str, int, float])

Return type
Union[Awaitable[Any], Any]

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)ret1 = r.setnx('name', 'yongqiang')
print(f"ret1: {ret1}")ret2 = r.setnx('name', 'cheng')
print(f"ret2: {ret2}")time.sleep(2)  # secs

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
ret1: True
ret2: FalseProcess finished with exit code 0

4.1.2. setex(name, time, value)

Set the value of key name to value that expires in time seconds. time can be represented by an integer or a Python timedelta object.

https://redis.io/docs/latest/commands/setex/

Parameters
name (Union[bytes, str, memoryview])
time (Union[int, timedelta])
value (Union[bytes, bytearray, memoryview, str, int, float])

Return type
Union[Awaitable[Any], Any]

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)r.setex('name', 2, 'yongqiang')print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")print("Start: %s" % time.ctime())
time.sleep(5)  # secs
print("End: %s" % time.ctime())print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
type: <class 'str'>
name: yongqiang
Start: Fri Aug 15 20:27:53 2025
End: Fri Aug 15 20:27:58 2025
name: NoneProcess finished with exit code 0

4.1.3. psetex(name, time_ms, value)

Set the value of key name to value that expires in time_ms milliseconds. time_ms can be represented by an integer or a Python timedelta object.

https://redis.io/commands/psetex

Parameters
name (Union[bytes, str, memoryview])
time_ms (Union[int, timedelta])
value (Union[bytes, bytearray, memoryview, str, int, float])

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)r.psetex('name', 2000, 'yongqiang')print(f"type: {type(r.get('name'))}")
print(f"name: {r.get('name')}")print("Start: %s" % time.ctime())
time.sleep(5)  # secs
print("End: %s" % time.ctime())print(f"name: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
type: <class 'str'>
name: yongqiang
Start: Fri Aug 15 20:32:44 2025
End: Fri Aug 15 20:32:49 2025
name: NoneProcess finished with exit code 0

4.2. flushdb(asynchronous=False, **kwargs)

Delete all keys in the current database.

https://redis.io/docs/latest/commands/flushdb/

asynchronous indicates whether the operation is executed asynchronously by the server.
清空當前 db 中的數據，默認是同步，異步 asynchronous=True，會新起一個線程進行清空操作，不阻塞主線程。

Parameters
asynchronous (bool)

Return type
Union[Awaitable[Any], Any]

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)

4.3. mset(mapping)

Sets key/values based on a mapping. Mapping is a dictionary of key/value pairs. Both keys and values should be strings or types that can be cast to a string via str().

https://redis.io/commands/mset

Parameters
mapping (Mapping[AnyKeyT, Union[bytes, bytearray, memoryview, str, int, float]])

Return type
Union[Awaitable[Any], Any]

4.4. mget(keys, *args)

Returns a list of values ordered identically to keys.

https://redis.io/commands/mget

Parameters
keys (Union[bytes, str, memoryview, Iterable[Union[bytes, str, memoryview]]])
args (Union[bytes, bytearray, memoryview, str, int, float])

Return type
Union[Awaitable[Any], Any]

#!/usr/bin/env python
# coding=utf-8import redis
import timepool = redis.ConnectionPool(host='localhost', port=6379, decode_responses=True)
r = redis.Redis(connection_pool=pool)
r.flushdb(asynchronous=False)r.mset({'k1': 'value1', 'k2': 'value2', 'k3': 'value3'})
print(r.mget("k1"))
print(r.mget("k1", "k2"))
print(r.mget("k1", "k2", "k3"))print(f"\nname: {r.get('name')}")

/home/yongqiang/miniconda3/bin/python /home/yongqiang/stable_diffusion_work/stable_diffusion_diffusers/forever.py 
['value1']
['value1', 'value2']
['value1', 'value2', 'value3']name: NoneProcess finished with exit code 0

References

[1] Yongqiang Cheng, https://yongqiang.blog.csdn.net/
[2] What is Redis? https://www.ibm.com/think/topics/redis