# Redis OM Python
了解如何使用 Redis Stack 和 Python 进行构建
Redis OM Python
是一个 Redis 客户端,它提供了用于管理 Redis 中的文档数据的高级抽象。本教程向您展示如何启动和运行 Redis OM Python、Redis Stack 和 Flask
微框架。
我们很乐意看到您使用 Redis Stack 和 Redis OM 构建的内容。 加入 Discord 上的 Redis 社区,
与我们讨论 Redis OM 和 Redis Stack 的所有内容。 在我们的公告博客文章
中阅读有关 Redis OM Python的更多信息。
# 概述
这个应用程序是一个使用 Flask 和一个简单的域模型构建的 API,演示了使用 Redis OM 的常见数据操作模式。
我们的实体是一个人,具有以下 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 的生成,它使用 ULIDs
来完成。Redis OM 还将为我们创建唯一的 Redis 键名,以及从存储在 Redis Stack 数据库中的 JSON 文档中保存和检索实体。
# 入门
# 要求
要运行此应用程序,您需要:
git
- 将 repo 克隆到您的机器上。Python 3.9 或更高版本
。Redis Stack
数据库,或安装了RediSearch
和RedisJSON
模块的Redis。我们docker-compose.yml
为此提供了一个。您还可以使用 Redis Enterprise Cloud 注册一个免费的 30Mb 数据库
- 请务必在创建云数据库时检查 Redis Stack 选项。curl
或Postman
- 向应用程序发送 HTTP 请求。我们将在本文档中提供使用 curl 的示例。- 可选:
RedisInsight
,一个免费的 Redis 数据可视化和数据库管理工具。下载RedisInsight的时候一定要选择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 Enterprise 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 Enterprise Cloud,则需要数据库的主机名、端口号和密码。使用这些来设置REDIS_OM_URL
环境变量,如下所示:
$ export REDIS_OM_URL=redis://default:<password>@<host>:<port>
(使用 Docker 时不需要此步骤,因为 Docker 容器在localhost
端口6379
上运行 Redis,无需密码,这是 Redis OM 使用的默认连接。)
例如,如果您的 Redis Enterprise Cloud 数据库位于9139
主机端口enterprise.redis.com
并且您的密码是,5uper53cret
那么您将设置REDIS_OM_URL
如下:
$ export REDIS_OM_URL=redis://default:5uper53cret@enterprise.redis.com:9139
# 创建 Python 虚拟环境并安装依赖项
创建 Python 虚拟环境,并安装项目依赖项,即 Flask
、 Requests
(仅在数据加载器脚本中使用)和 Redis OM
:
$ python3 -m venv venv
$ . ./venv/bin/activate
$ pip install -r requirements.txt
# 启动 Flask 应用程序
让我们以开发模式启动 Flask 应用程序,这样每次您保存代码更改时,Flask 都会为您重新启动服务器app.py
:
$ 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、RediSearch、RedisJSON 和 Redis OM for Python 对数据执行 CRUD 操作!为确保服务器正在运行,请将浏览器指向http://127.0.0.1:5000/
,您可以在此处看到应用程序的基本主页:
# 加载样本数据
我们提供了少量示例数据(它位于data/people.json
.Python 脚本中dataloader.py
,通过将数据发布到应用程序的 create a new person 端点来将每个人加载到 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_OM_URL
如果使用 Redis Enterprise Cloud,则需要设置环境变量。
如果您设置了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 构建人物模型
Redis OM 允许我们使用 Python 类和 Pydantic
框架对实体进行建模。我们的人模型包含在文件中person.py
。以下是有关其工作原理的一些说明:
- 我们声明了一个
Person
扩展 Redis OM 类的类JsonModel
。这告诉 Redis OM 我们希望将这些实体作为 JSON 文档存储在 Redis 中。 - 然后,我们在模型中声明每个字段,指定数据类型以及是否要在该字段上建立索引。例如,这里是
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 表示中,地址是一个对象,而不是字符串或数字字段。对于 Redis OM,这被建模为第二个类,它扩展了 Redis OMEmbeddedJsonModel
类:
class Address(EmbeddedJsonModel):
# field definitions...
- an 中的字段以
EmbeddedJsonModel
相同的方式定义,因此我们的类包含地址中每个数据项的字段定义。 - 并非 JSON 中的每个字段都存在于每个地址中,Redis OM 允许我们将一个字段声明为可选,只要我们不对其进行索引:
unit: Optional[str] = Field(index=False)
- 我们还可以为字段设置默认值......假设国家应该是“英国”,除非另有说明:
country: str = Field(index=True, default="United Kingdom")
- 最后,为了将嵌入的地址对象添加到我们的 Person 模型中,我们
Address
在 Person 类中声明了一个类型字段:
address: Address
# 添加新人
create_person
中的函数app.py
处理 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 中保存的内容。使用 RedisInsight 或 redis-cli,连接到数据库并查看存储在 key 中的值:person.Person:01FX8SSSDN7PT9T3N0JZZA758G
。这在 Redis 中存储为 JSON 文档,因此如果使用 redis-cli,您将需要以下命令:
$ redis-cli
127.0.0.1:6379> json.get :person.Person:01FX8SSSDN7PT9T3N0JZZA758G
如果您使用的是 RedisInsight,当您单击键名时,浏览器将为您呈现键值:
当在 Redis 中以 JSON 格式存储数据时,我们可以更新和检索整个文档,或者只是其中的一部分。例如,要仅检索人员的地址和第一项技能,请使用以下命令(RedisInsight 用户应为此使用内置的 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 路径语法的更多信息,请参阅 RedisJSON 文档
。
# 通过 ID 查找人员
如果我们知道一个人的 ID,我们就可以检索他们的数据。find_by_id
中的函数app.py
接收一个 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"
]
}
# 查找名字和姓氏匹配的人
让我们找到所有具有给定名字和姓氏的人......这由find_by_name
.app.py
在这里,我们使用find
Redis OM 提供的 Person 类方法。我们向它传递一个搜索查询,指定我们要查找其first_name
字段包含first_name
传递给的参数值的人,以及find_by_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"
]
}
]
}
# 查找给定年龄范围内的人
能够找到属于给定年龄范围的人很有用...中的函数find_in_age_range
处理app.py
如下...
我们将再次使用 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
技能数组包含的人desired_skill
”,其中city
和desired_skill
是find_matching_skill
函数的参数app.py
。这是代码:
people = Person.find(
(Person.skills << desired_skill) &
(Person.address.city == city)
).all()
这里的<<
操作符用来表示“in”或“contains”。
让我们找到谢菲尔德的所有吉他手:
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
字段,它是一个包含关于他们的几句话的自由文本字符串。我们选择以一种使其全文可搜索的方式对其进行索引,所以现在让我们看看如何使用它。这个的代码在函数find_matching_statements
中app.py
。
要搜索search_term
在其personal_statement
字段中具有参数值的人,我们使用%
运算符:
Person.find(Person.personal_statement % search_term).all()
让我们找到每个在个人陈述中谈论“玩”的人。
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 中检索信息外,我们还需要不时更新 Person 的数据。让我们看看如何使用 Redis OM for Python 做到这一点。
update_age
中的函数app.py
接受两个参数: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 对象中。在 中的函数delete_person
中app.py
,我们调用delete
Person 类的类方法来执行此操作:
Person.delete(id)
让我们删除 ID 为 Dan Harris 的人01FX8RMR8545RWW4DYCE5MSZA1
:
curl --location --request POST 'http://127.0.0.1:5000/person/01FX8RMR8545RWW4DYCE5MSZA1/delete'
ok
无论提供的 ID 是否存在于 Redis 中,服务器都会响应。
# 为人员设置到期时间
这是一个如何对保存在 Redis 中的模型实例运行任意 Redis 命令的示例。让我们看看如何设置一个人的生存时间 (TTL),以便 Redis 在经过可配置的秒数后使 JSON 文档过期。
中的函数expire_by_id
按app.py
如下方式处理此问题。它需要两个参数:id
- 要过期的人的 ID,以及seconds
- 未来要过期的人的秒数。这需要我们 EXPIRE
针对该人的密钥运行 Redis 命令。为此,我们需要从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命令redis-cli
检查此人现在是否设置了 TTL :expire
127.0.0.1:6379> ttl :person.Person:01FX8RMR82D091TC37B45RCWY3
(integer) 584
这表明 Redis 将在 584 秒后过期密钥。
.db()
每当您想运行较低级别的 Redis 命令时,您可以使用模型类上的函数来获取底层 redis-py 连接。有关更多详细信息,请参阅 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
← Node.js SpringOrJava →