RedisOM for Python
了解如何使用 Redis Stack 和 Python 进行构建
Redis OM Python 是一个 Redis 客户端,提供了用于管理 Redis 中文档数据的高级抽象。本教程将向您展示如何开始使用 Redis OM Python、Redis Stack 和 Flask 微框架。
我们很想看看您使用 Redis Stack 和 Redis OM 构建了什么。加入 Redis 社区 Discord 频道,与我们交流有关 Redis OM 和 Redis Stack 的所有内容。通过我们的公告博客文章了解更多关于 Redis OM Python 的信息。
概览
此应用程序是一个使用 Flask 构建的 API 和一个简单的领域模型,演示了使用 Redis OM 的常见数据操作模式。
我们的实体是一个 Person(人员),其 JSON 表示如下
{
"first_name": "A string, the person's first or given name",
"last_name": "A string, the person's last or surname",
"age": 36,
"address": {
"street_number": 56,
"unit": "A string, optional unit number e.g. B or 1",
"street_name": "A string, name of the street they live on",
"city": "A string, name of the city they live in",
"state": "A string, state, province or county that they live in",
"postal_code": "A string, their zip or postal code",
"country": "A string, country that they live in."
},
"personal_statement": "A string, free text personal statement",
"skills": [
"A string: a skill the person has",
"A string: another still that the person has"
]
}
我们将让 Redis OM 处理唯一 ID 的生成,它使用 ULID 来实现。Redis OM 还将为我们处理唯一的 Redis 键名创建,以及从存储在 Redis Stack 数据库中的 JSON 文档中保存和检索实体。
入门
要求
要运行此应用程序,您需要
- git - 将仓库克隆到您的机器上。
- Python 3.9 或更高版本.
- 一个 Redis Stack 数据库,或者安装了搜索与查询和JSON功能的 Redis。我们为此提供了一个
docker-compose.yml
。您还可以注册一个免费的 Redis Cloud 30MB 数据库 - 在创建云数据库时务必勾选 Redis Stack 选项。 - curl 或 Postman - 用于向应用程序发送 HTTP 请求。本文档中将提供使用 curl 的示例。
- 可选:Redis Insight,一个免费的 Redis 数据可视化和数据库管理工具。下载 Redis Insight 时,请务必选择 2.x 版本或使用 Redis Stack 附带的版本。
获取源代码
从 GitHub 克隆仓库
$ git clone https://github.com/redis-developer/redis-om-python-flask-skeleton-app.git
$ cd redis-om-python-flask-skeleton-app
启动 Redis Stack 数据库,或配置您的 Redis Cloud 凭据
接下来,我们将启动并运行一个 Redis Stack 数据库。如果您使用 Docker
$ docker-compose up -d
Creating network "redis-om-python-flask-skeleton-app_default" with the default driver
Creating redis_om_python_flask_starter ... done
如果您使用 Redis Cloud,则需要数据库的主机名、端口号和密码。使用这些信息像这样设置 REDIS_OM_URL
环境变量
$ export REDIS_OM_URL=redis://default:<password>@<host>:<port>
(如果使用 Docker,则不需要此步骤,因为 Docker 容器在 localhost
的 6379
端口上运行 Redis,且没有密码,这是 Redis OM 使用的默认连接。)
例如,如果您的 Redis Cloud 数据库位于主机 enterprise.redis.com
的 9139
端口,并且您的密码是 5uper53cret
,那么您将按如下方式设置 REDIS_OM_URL
$ export REDIS_OM_URL=redis://default:[email protected]:9139
创建 Python 虚拟环境并安装依赖项
创建一个 Python 虚拟环境,并安装项目依赖项,包括 Flask、Requests(仅在数据加载脚本中使用)和 Redis OM
$ python3 -m venv venv
$ . ./venv/bin/activate
$ pip install -r requirements.txt
启动 Flask 应用程序
让我们在开发模式下启动 Flask 应用程序,这样每次您在 app.py
中保存代码更改时,Flask 都会为您重启服务器
$ export FLASK_ENV=development
$ flask run
如果一切顺利,您应该会看到类似以下的输出
$ flask run
* Environment: development
* Debug mode: on
* Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
* Restarting with stat
* Debugger is active!
* Debugger PIN: XXX-XXX-XXX
您现在已经启动并运行,准备好使用 Redis、搜索与查询、JSON 和 Redis OM for Python 对数据执行 CRUD 操作了!为了确保服务器正在运行,请将浏览器指向 http://127.0.0.1:5000/
,您应该会看到应用程序的基本主页
加载示例数据
我们提供了一些示例数据(在 data/people.json
中)。Python 脚本 dataloader.py
通过将数据发布到应用程序的创建新人员端点来将每个人员加载到 Redis 中。按如下方式运行它
$ python dataloader.py
Created person Robert McDonald with ID 01FX8RMR7NRS45PBT3XP9KNAZH
Created person Kareem Khan with ID 01FX8RMR7T60ANQTS4P9NKPKX8
Created person Fernando Ortega with ID 01FX8RMR7YB283BPZ88HAG066P
Created person Noor Vasan with ID 01FX8RMR82D091TC37B45RCWY3
Created person Dan Harris with ID 01FX8RMR8545RWW4DYCE5MSZA1
务必复制数据加载器的输出,因为您的 ID 将与教程中使用的不同。为了跟随教程,请用您自己的 ID 替换上面显示的 ID。例如,每当我们使用 Kareem Khan 时,将 01FX8RMR7T60ANQTS4P9NKPKX8
替换为您的数据加载器在您的 Redis 数据库中分配给 Kareem 的 ID。
问题?
如果 Flask 服务器启动失败,请查看其输出。如果您看到类似这样的日志条目
raise ConnectionError(self._error_message(e))
redis.exceptions.ConnectionError: Error 61 connecting to localhost:6379. Connection refused.
那么如果使用 Docker,您需要启动 Redis Docker 容器;如果使用 Redis Cloud,则需要设置 REDIS_OM_URL
环境变量。
如果您已经设置了 REDIS_OM_URL
环境变量,但代码在启动时出现类似这样的错误
raise ConnectionError(self._error_message(e))
redis.exceptions.ConnectionError: Error 8 connecting to enterprise.redis.com:9139. nodename nor servname provided, or not known.
那么您需要检查设置 REDIS_OM_URL
时使用的主机名、端口、密码和格式是否正确。
如果数据加载器未能将示例数据发布到应用程序中,请确保在运行数据加载器之前 Flask 应用程序已经运行。
创建、读取、更新和删除数据
让我们在 Redis 中创建和操作数据模型的一些实例。这里我们将介绍如何使用 curl(您也可以使用 Postman)调用 Flask API,代码如何工作,以及数据如何存储在 Redis 中。
使用 Redis OM 构建 Person 模型
Redis OM 允许我们使用 Python 类和 Pydantic 框架来建模实体。我们的人员模型包含在 person.py
文件中。以下是关于其工作原理的一些注意事项
- 我们声明了一个名为
Person
的类,它继承自 Redis OM 类JsonModel
。这告诉 Redis OM 我们希望将这些实体存储在 Redis 中作为 JSON 文档。 - 然后,我们声明模型中的每个字段,指定数据类型以及是否希望在该字段上创建索引。例如,这是
age
字段,我们将其声明为一个正整数,并希望在其上创建索引
age: PositiveInt = Field(index=True)
skills
字段是一个字符串列表,声明如下
skills: List[str] = Field(index=True)
- 对于
personal_statement
字段,我们不希望对字段的值创建索引,因为它是一个自由文本句子,而不是单个单词或数字。为此,我们将告诉 Redis OM 我们希望能够对值执行全文搜索
personal_statement: str = Field(index=True, full_text_search=True)
address
与其他字段的工作方式不同。请注意,在模型的 JSON 表示中,address
是一个对象而不是字符串或数字字段。使用 Redis OM,这被建模为第二个类,它继承自 Redis OM 的EmbeddedJsonModel
类
class Address(EmbeddedJsonModel):
# field definitions...
-
EmbeddedJsonModel
中的字段定义方式相同,因此我们的类包含地址中每个数据项的字段定义。 -
并非 JSON 中的每个字段都存在于每个地址中,Redis OM 允许我们将字段声明为可选,只要我们不对其创建索引
unit: Optional[str] = Field(index=False)
- 我们还可以为字段设置默认值... 例如,除非另有指定,否则国家应为“United Kingdom”
country: str = Field(index=True, default="United Kingdom")
- 最后,要将嵌入的
address
对象添加到我们的Person
模型中,我们在Person
类中声明一个类型为Address
的字段
address: Address
添加新人员
app.py
中的 create_person
函数负责在 Redis 中创建新人员。它期望一个符合我们 Person 模型模式的 JSON 对象。然后使用该数据创建新的 Person
对象并将其保存在 Redis 中的代码很简单
new_person = Person(**request.json)
new_person.save()
return new_person.pk
创建新的 Person
实例时,Redis OM 会为其分配一个唯一的 ULID 主键,我们可以通过 .pk
访问它。我们将其返回给调用者,以便他们知道刚刚创建的对象的 ID。
然后将对象持久化到 Redis 中只需调用其 .save()
方法。
试试看... 在服务器运行的情况下,使用 curl 添加新人员
curl --location --request POST 'http://127.0.0.1:5000/person/new' \
--header 'Content-Type: application/json' \
--data-raw '{
"first_name": "Joanne",
"last_name": "Peel",
"age": 36,
"personal_statement": "Music is my life, I love gigging and playing with my band.",
"address": {
"street_number": 56,
"unit": "4A",
"street_name": "The Rushes",
"city": "Birmingham",
"state": "West Midlands",
"postal_code": "B91 6HG",
"country": "United Kingdom"
},
"skills": [
"synths",
"vocals",
"guitar"
]
}'
运行上面的 curl 命令将返回分配给新创建人员的唯一 ULID ID。例如 01FX8SSSDN7PT9T3N0JZZA758G
。
检查 Redis 中的数据
让我们看看刚刚在 Redis 中保存了什么。使用 Redis Insight 或 redis-cli,连接到数据库并查看键 :person.Person:01FX8SSSDN7PT9T3N0JZZA758G
处存储的值。这在 Redis 中存储为 JSON 文档,因此如果使用 redis-cli,您需要使用以下命令
$ redis-cli
127.0.0.1:6379> json.get :person.Person:01FX8SSSDN7PT9T3N0JZZA758G
如果您使用 Redis Insight,单击键名时浏览器会为您渲染键值
当在 Redis 中将数据存储为 JSON 时,我们可以更新和检索整个文档,或者只检索其中的一部分。例如,要仅检索人员的地址和第一个技能,请使用以下命令(Redis Insight 用户应使用内置的 redis-cli)
$ redis-cli
127.0.0.1:6379> json.get :person.Person:01FX8SSSDN7PT9T3N0JZZA758G $.address $.skills[0]
"{\"$.skills[0]\":[\"synths\"],\"$.address\":[{\"pk\":\"01FX8SSSDNRDSRB3HMVH00NQTT\",\"street_number\":56,\"unit\":\"4A\",\"street_name\":\"The Rushes\",\"city\":\"Birmingham\",\"state\":\"West Midlands\",\"postal_code\":\"B91 6HG\",\"country\":\"United Kingdom\"}]}"
有关在 Redis 中查询 JSON 文档所使用的 JSON Path 语法的更多信息,请参阅文档。
按 ID 查找人员
如果我们知道人员的 ID,就可以检索他们的数据。app.py
中的 find_by_id
函数接收 ID 作为参数,并要求 Redis OM 使用该 ID 和 Person
类的 .get
方法来检索并填充一个 Person
对象
try:
person = Person.get(id)
return person.dict()
except NotFoundError:
return {}
.dict()
方法将我们的 Person
对象转换为 Python 字典,然后 Flask 将其返回给调用者。
请注意,如果 Redis 中没有具有提供的 ID 的 Person,get
将抛出 NotFoundError
。
使用 curl 试试这个,将 01FX8SSSDN7PT9T3N0JZZA758G
替换为您刚刚在数据库中创建的人员的 ID
curl --location --request GET 'http://localhost:5000/person/byid/01FX8SSSDN7PT9T3N0JZZA758G'
服务器响应一个包含用户数据的 JSON 对象
{
"address": {
"city": "Birmingham",
"country": "United Kingdom",
"pk": "01FX8SSSDNRDSRB3HMVH00NQTT",
"postal_code": "B91 6HG",
"state": "West Midlands",
"street_name": "The Rushes",
"street_number": 56,
"unit": null
},
"age": 36,
"first_name": "Joanne",
"last_name": "Peel",
"personal_statement": "Music is my life, I love gigging and playing with my band.",
"pk": "01FX8SSSDN7PT9T3N0JZZA758G",
"skills": [
"synths",
"vocals",
"guitar"
]
}
按名字和姓氏查找匹配人员
让我们查找所有具有给定名字和姓氏的人员... 这由 app.py
中的 find_by_name
函数处理。
这里,我们使用 Redis OM 提供的 Person
类的 find
方法。我们向其传递一个搜索查询,指定我们希望找到 first_name
字段包含传递给 find_by_name
的 first_name
参数值并且 last_name
字段包含 last_name
参数值的人员
people = Person.find(
(Person.first_name == first_name) &
(Person.last_name == last_name)
).all()
.all()
告诉 Redis OM 我们希望检索所有匹配的人员。
使用 curl 按如下方式试试看
curl --location --request GET 'http://127.0.0.1:5000/people/byname/Kareem/Khan'
注意:名字和姓氏区分大小写。
服务器响应一个包含 results
对象,其中是一个匹配人员的数组
{
"results": [
{
"address": {
"city": "Sheffield",
"country": "United Kingdom",
"pk": "01FX8RMR7THMGA84RH8ZRQRRP9",
"postal_code": "S1 5RE",
"state": "South Yorkshire",
"street_name": "The Beltway",
"street_number": 1,
"unit": "A"
},
"age": 27,
"first_name": "Kareem",
"last_name": "Khan",
"personal_statement":"I'm Kareem, a multi-instrumentalist and singer looking to join a new rock band.",
"pk":"01FX8RMR7T60ANQTS4P9NKPKX8",
"skills": [
"drums",
"guitar",
"synths"
]
}
]
}
查找给定年龄范围内的人员
能够找到属于给定年龄范围的人员非常有用... app.py
中的 find_in_age_range
函数按如下方式处理此问题...
我们将再次使用 Person
类的 find
方法,这次向其传递一个最小和最大年龄,指定我们只希望查找 age
字段介于这些值之间的人员
people = Person.find(
(Person.age >= min_age) &
(Person.age <= max_age)
).sort_by("age").all()
请注意,我们还可以使用 .sort_by
来指定我们希望结果按哪个字段排序。
让我们查找所有年龄在 30 到 47 岁之间的人员,并按年龄排序
curl --location --request GET 'http://127.0.0.1:5000/people/byage/30/47'
这将返回一个包含 results
对象的数组,其中是匹配人员
{
"results": [
{
"address": {
"city": "Sheffield",
"country": "United Kingdom",
"pk": "01FX8RMR7NW221STN6NVRDPEDT",
"postal_code": "S12 2MX",
"state": "South Yorkshire",
"street_name": "Main Street",
"street_number": 9,
"unit": null
},
"age": 35,
"first_name": "Robert",
"last_name": "McDonald",
"personal_statement": "My name is Robert, I love meeting new people and enjoy music, coding and walking my dog.",
"pk": "01FX8RMR7NRS45PBT3XP9KNAZH",
"skills": [
"guitar",
"piano",
"trombone"
]
},
{
"address": {
"city": "Birmingham",
"country": "United Kingdom",
"pk": "01FX8SSSDNRDSRB3HMVH00NQTT",
"postal_code": "B91 6HG",
"state": "West Midlands",
"street_name": "The Rushes",
"street_number": 56,
"unit": null
},
"age": 36,
"first_name": "Joanne",
"last_name": "Peel",
"personal_statement": "Music is my life, I love gigging and playing with my band.",
"pk": "01FX8SSSDN7PT9T3N0JZZA758G",
"skills": [
"synths",
"vocals",
"guitar"
]
},
{
"address": {
"city": "Nottingham",
"country": "United Kingdom",
"pk": "01FX8RMR82DDJ90CW8D1GM68YZ",
"postal_code": "NG1 1AA",
"state": "Nottinghamshire",
"street_name": "Broadway",
"street_number": 12,
"unit": "A-1"
},
"age": 37,
"first_name": "Noor",
"last_name": "Vasan",
"personal_statement": "I sing and play the guitar, I enjoy touring and meeting new people on the road.",
"pk": "01FX8RMR82D091TC37B45RCWY3",
"skills": [
"vocals",
"guitar"
]
},
{
"address": {
"city": "San Diego",
"country": "United States",
"pk": "01FX8RMR7YCDAVSWBMWCH2B07G",
"postal_code": "92102",
"state": "California",
"street_name": "C Street",
"street_number": 1299,
"unit": null
},
"age": 43,
"first_name": "Fernando",
"last_name": "Ortega",
"personal_statement": "I'm in a really cool band that plays a lot of cover songs. I'm the drummer!",
"pk": "01FX8RMR7YB283BPZ88HAG066P",
"skills": [
"clarinet",
"oboe",
"drums"
]
}
]
}
在给定城市中查找具有特定技能的人员
现在,我们将尝试一种稍微不同的查询。我们希望找到所有居住在给定城市并且具有某种特定技能的人员。这需要在作为字符串的 city
字段和作为字符串数组的 skills
字段上都进行搜索。
本质上,我们希望表达“找到所有城市是 city
并且 skills
数组包含 desired_skill
的人员”,其中 city
和 desired_skill
是 app.py
中 find_matching_skill
函数的参数。代码如下
people = Person.find(
(Person.skills << desired_skill) &
(Person.address.city == city)
).all()
这里的 <<
运算符用于表示“在其中”或“包含”。
让我们查找 Sheffield 的所有吉他手
curl --location --request GET 'http://127.0.0.1:5000/people/byskill/guitar/Sheffield'
注意:Sheffield
区分大小写。
服务器返回一个包含匹配人员的 results
数组
{
"results": [
{
"address": {
"city": "Sheffield",
"country": "United Kingdom",
"pk": "01FX8RMR7THMGA84RH8ZRQRRP9",
"postal_code": "S1 5RE",
"state": "South Yorkshire",
"street_name": "The Beltway",
"street_number": 1,
"unit": "A"
},
"age": 28,
"first_name": "Kareem",
"last_name": "Khan",
"personal_statement": "I'm Kareem, a multi-instrumentalist and singer looking to join a new rock band.",
"pk": "01FX8RMR7T60ANQTS4P9NKPKX8",
"skills": [
"drums",
"guitar",
"synths"
]
},
{
"address": {
"city": "Sheffield",
"country": "United Kingdom",
"pk": "01FX8RMR7NW221STN6NVRDPEDT",
"postal_code": "S12 2MX",
"state": "South Yorkshire",
"street_name": "Main Street",
"street_number": 9,
"unit": null
},
"age": 35,
"first_name": "Robert",
"last_name": "McDonald",
"personal_statement": "My name is Robert, I love meeting new people and enjoy music, coding and walking my dog.",
"pk": "01FX8RMR7NRS45PBT3XP9KNAZH",
"skills": [
"guitar",
"piano",
"trombone"
]
}
]
}
使用全文搜索在人员个人陈述中查找人员
每个人员都有一个 personal_statement
字段,这是一个自由文本字符串,包含几句关于他们的陈述。我们选择以一种支持全文搜索的方式对该字段创建索引,因此现在让我们看看如何使用它。相关代码位于 app.py
中的 find_matching_statements
函数中。
要在人员的 personal_statement
字段中搜索包含参数 search_term
值的人员,我们使用 %
运算符
Person.find(Person.personal_statement % search_term).all()
让我们查找所有在个人陈述中提到“play”的人员。
curl --location --request GET 'http://127.0.0.1:5000/people/bystatement/play'
服务器响应一个包含匹配人员的 results
数组
{
"results": [
{
"address": {
"city": "San Diego",
"country": "United States",
"pk": "01FX8RMR7YCDAVSWBMWCH2B07G",
"postal_code": "92102",
"state": "California",
"street_name": "C Street",
"street_number": 1299,
"unit": null
},
"age": 43,
"first_name": "Fernando",
"last_name": "Ortega",
"personal_statement": "I'm in a really cool band that plays a lot of cover songs. I'm the drummer!",
"pk": "01FX8RMR7YB283BPZ88HAG066P",
"skills": [
"clarinet",
"oboe",
"drums"
]
}, {
"address": {
"city": "Nottingham",
"country": "United Kingdom",
"pk": "01FX8RMR82DDJ90CW8D1GM68YZ",
"postal_code": "NG1 1AA",
"state": "Nottinghamshire",
"street_name": "Broadway",
"street_number": 12,
"unit": "A-1"
},
"age": 37,
"first_name": "Noor",
"last_name": "Vasan",
"personal_statement": "I sing and play the guitar, I enjoy touring and meeting new people on the road.",
"pk": "01FX8RMR82D091TC37B45RCWY3",
"skills": [
"vocals",
"guitar"
]
},
{
"address": {
"city": "Birmingham",
"country": "United Kingdom",
"pk": "01FX8SSSDNRDSRB3HMVH00NQTT",
"postal_code": "B91 6HG",
"state": "West Midlands",
"street_name": "The Rushes",
"street_number": 56,
"unit": null
},
"age": 36,
"first_name": "Joanne",
"last_name": "Peel",
"personal_statement": "Music is my life, I love gigging and playing with my band.",
"pk": "01FX8SSSDN7PT9T3N0JZZA758G",
"skills": [
"synths",
"vocals",
"guitar"
]
}
]
}
请注意,我们获得的结果包括匹配“play”、“plays”和“playing”。
更新人员年龄
除了从 Redis 中检索信息,我们有时还需要更新人员数据。让我们看看如何使用 Redis OM for Python 来实现这一点。
app.py
中的 update_age
函数接受两个参数:id
和 new_age
。使用这些参数,我们首先从 Redis 中检索人员数据并用其创建一个新对象
try:
person = Person.get(id)
except NotFoundError:
return "Bad request", 400
假设我们找到了人员,让我们更新他们的年龄并将数据保存回 Redis
person.age = new_age
person.save()
让我们将 Kareem Khan 的年龄从 27 岁更改为 28 岁
curl --location --request POST 'http://127.0.0.1:5000/person/01FX8RMR7T60ANQTS4P9NKPKX8/age/28'
服务器响应 ok
。
删除人员
如果我们知道人员的 ID,就可以直接从 Redis 中删除他们,而无需先将其数据加载到 Person
对象中。在 app.py
中的 delete_person
函数中,我们调用 Person
类的 delete
方法来完成此操作
Person.delete(id)
让我们删除 Dan Harris,其 ID 为 01FX8RMR8545RWW4DYCE5MSZA1
curl --location --request POST 'http://127.0.0.1:5000/person/01FX8RMR8545RWW4DYCE5MSZA1/delete'
无论提供的 ID 是否存在于 Redis 中,服务器都将响应 ok
。
为人员设置过期时间
这是一个如何在 Redis 中对模型实例运行任意 Redis 命令的示例。让我们看看如何为人员设置生存时间 (TTL),以便 Redis 在可配置的秒数后使 JSON 文档过期。
app.py
中的 expire_by_id
函数按如下方式处理此问题。它接受两个参数:id
- 要过期的人员 ID,以及 seconds
- 在未来多少秒后过期人员。这需要我们针对人员的键运行 Redis 的 EXPIRE
命令。为此,我们需要从 Person
模型中访问 Redis 连接,如下所示
person_to_expire = Person.get(id)
Person.db().expire(person_to_expire.key(), seconds)
让我们将 ID 为 01FX8RMR82D091TC37B45RCWY3
的人员设置为在 600 秒后过期
curl --location --request POST 'http://localhost:5000/person/01FX8RMR82D091TC37B45RCWY3/expire/600'
使用 redis-cli
,您可以使用 Redis expire
命令检查该人员是否已设置 TTL
127.0.0.1:6379> ttl :person.Person:01FX8RMR82D091TC37B45RCWY3
(integer) 584
这表明 Redis 将在 584 秒后使该键过期。
您可以随时在模型类上使用 .db()
函数来获取底层的 redis-py 连接,以便运行更低级别的 Redis 命令。有关更多详细信息,请参阅 redis-py 文档。
关闭 Redis (Docker)
如果您使用 Docker,并且在完成应用程序后希望关闭 Redis 容器,请使用 docker-compose down
$ docker-compose down
Stopping redis_om_python_flask_starter ... done
Removing redis_om_python_flask_starter ... done
Removing network redis-om-python-flask-skeleton-app_default