Python 指南
将您的 Python 应用程序连接到 Redis 数据库
redis-py 是 Redis 的 Python 客户端。以下部分说明如何安装 redis-py 并将您的应用程序连接到 Redis 数据库。
redis-py 需要运行的 Redis 或 Redis Stack 服务器。请参阅 入门 获取 Redis 安装说明。
您还可以使用对象映射客户端接口访问 Redis。请参阅 RedisOM for Python 获取更多信息。
安装
要安装 redis-py，请输入
pip install redis
为了获得更快的性能，请使用 hiredis 支持安装 Redis。这将提供一个已编译的响应解析器，并且在大多数情况下无需进行任何代码更改。默认情况下，如果 hiredis >= 1.0 可用，redis-py 会尝试使用它进行响应解析。
distutils 打包方案不再是 Python 3.12 及更高版本的组成部分。如果您在 Python 3.12 环境中安装 redis-py 时遇到困难，请考虑更新到 redis-py 的最新版本。pip install redis[hiredis]
连接
连接到端口 6379 上的本地主机，在 Redis 中设置值，并检索它。所有响应都以 Python 中的字节形式返回。要接收解码的字符串，请设置 decode_responses=True。有关更多连接选项，请参阅 这些示例。
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
存储和检索一个简单的字符串。
r.set('foo', 'bar')
# True
r.get('foo')
# bar
存储和检索一个字典。
r.hset('user-session:123', mapping={
'name': 'John',
"surname": 'Smith',
"company": 'Redis',
"age": 29
})
# True
r.hgetall('user-session:123')
# {'surname': 'Smith', 'name': 'John', 'company': 'Redis', 'age': '29'}
连接到 Redis 集群
要连接到 Redis 集群，请使用 RedisCluster。
from redis.cluster import RedisCluster
rc = RedisCluster(host='localhost', port=16379)
print(rc.get_nodes())
# [[host=127.0.0.1,port=16379,name=127.0.0.1:16379,server_type=primary,redis_connection=Redis<ConnectionPool<Connection<host=127.0.0.1,port=16379,db=0>>>], ...
rc.set('foo', 'bar')
# True
rc.get('foo')
# b'bar'
有关更多信息，请参阅 redis-py 集群。
使用 TLS 连接到您的生产 Redis
部署应用程序时，请使用 TLS 并遵循Redis 安全指南。
import redis
r = redis.Redis(
host="my-redis.cloud.redislabs.com", port=6379,
username="default", # use your Redis user. More info https://redis.ac.cn/docs/latest/operate/oss_and_stack/management/security/acl/
password="secret", # use your Redis password
ssl=True,
ssl_certfile="./redis_user.crt",
ssl_keyfile="./redis_user_private.key",
ssl_ca_certs="./redis_ca.pem",
)
r.set('foo', 'bar')
# True
r.get('foo')
# b'bar'
有关更多信息，请参阅redis-py TLS 示例。
示例：索引和查询 JSON 文档
确保您已安装 Redis Stack 和 redis-py。导入依赖项
import redis
from redis.commands.json.path import Path
import redis.commands.search.aggregation as aggregations
import redis.commands.search.reducers as reducers
from redis.commands.search.field import TextField, NumericField, TagField
from redis.commands.search.indexDefinition import IndexDefinition, IndexType
from redis.commands.search.query import NumericFilter, Query
连接到您的 Redis 数据库。
r = redis.Redis(host='localhost', port=6379)
让我们创建一些测试数据以添加到您的数据库中。
user1 = {
"name": "Paul John",
"email": "paul.john@example.com",
"age": 42,
"city": "London"
}
user2 = {
"name": "Eden Zamir",
"email": "eden.zamir@example.com",
"age": 29,
"city": "Tel Aviv"
}
user3 = {
"name": "Paul Zamir",
"email": "paul.zamir@example.com",
"age": 35,
"city": "Tel Aviv"
}
使用 schema 定义索引字段及其数据类型。使用 JSON 路径表达式将特定 JSON 元素映射到模式字段。
schema = (
TextField("$.name", as_name="name"),
TagField("$.city", as_name="city"),
NumericField("$.age", as_name="age")
)
创建索引。在本示例中，将索引所有具有键前缀 user: 的 JSON 文档。有关更多信息，请参阅查询语法。
rs = r.ft("idx:users")
rs.create_index(
schema,
definition=IndexDefinition(
prefix=["user:"], index_type=IndexType.JSON
)
)
# b'OK'
使用JSON.SET 在指定路径设置每个用户值。
r.json().set("user:1", Path.root_path(), user1)
r.json().set("user:2", Path.root_path(), user2)
r.json().set("user:3", Path.root_path(), user3)
让我们查找用户 Paul 并根据年龄过滤结果。
res = rs.search(
Query("Paul @age:[30 40]")
)
# Result{1 total, docs: [Document {'id': 'user:3', 'payload': None, 'json': '{"name":"Paul Zamir","email":"paul.zamir@example.com","age":35,"city":"Tel Aviv"}'}]}
使用 JSON 路径表达式查询。
rs.search(
Query("Paul").return_field("$.city", as_field="city")
).docs
# [Document {'id': 'user:1', 'payload': None, 'city': 'London'}, Document {'id': 'user:3', 'payload': None, 'city': 'Tel Aviv'}]
使用FT.AGGREGATE 聚合结果。
req = aggregations.AggregateRequest("*").group_by('@city', reducers.count().alias('count'))
print(rs.aggregate(req).rows)
# [[b'city', b'Tel Aviv', b'count', b'2'], [b'city', b'London', b'count', b'1']]