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 文档中保存和检索实体。

入门

要求

要运行此应用程序,您需要

获取源代码

从 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 容器在 localhost6379 端口上运行 Redis,且没有密码,这是 Redis OM 使用的默认连接。)

例如,如果您的 Redis Cloud 数据库位于主机 enterprise.redis.com9139 端口,并且您的密码是 5uper53cret,那么您将按如下方式设置 REDIS_OM_URL

$ export REDIS_OM_URL=redis://default:[email protected]:9139

创建 Python 虚拟环境并安装依赖项

创建一个 Python 虚拟环境,并安装项目依赖项,包括 FlaskRequests(仅在数据加载脚本中使用)和 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/,您应该会看到应用程序的基本主页

screenshot

加载示例数据

我们提供了一些示例数据(在 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,单击键名时浏览器会为您渲染键值

Data in 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_namefirst_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 的人员”,其中 citydesired_skillapp.pyfind_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 函数接受两个参数:idnew_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
评价此页面
回到顶部 ↑