Python 版 RedisOM

了解如何使用 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 的生成，它使用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 云凭证

接下来，我们将启动并运行 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端口上运行 Redis 6379，这是 Redis OM 使用的默认连接。）

例如，如果您的 Redis 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、搜索和查询、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 替换为上面显示的 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 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 调用 Flask API（您也可以使用 Postman）、代码如何工作以及数据如何存储在 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 OM 类EmbeddedJsonModel：
class Address(EmbeddedJsonModel):
    # field definitions...

  • 中的字段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 中保存的内容。使用 Redis Insight 或 redis-cli，连接到数据库并查看存储在 key 处的值:person.Person:01FX8SSSDN7PT9T3N0JZZA758G。这以 JSON 文档的形式存储在 Redis 中，因此如果使用 redis-cli，则需要以下命令：

$ redis-cli
127.0.0.1:6379> json.get :person.Person:01FX8SSSDN7PT9T3N0JZZA758G

如果您使用 Redis Insight，当您单击键名称时，浏览器将为您呈现键值：

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 路径语法的更多信息，请参阅文档。

通过 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。

这里我们使用findRedis OM 提供的 Person 类方法。我们向其传递一个搜索查询，指定我们想要查找字段包含传递给的参数first_name的值且字段包含该参数值的人员：first_namefind_by_namelast_namelast_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_skillapp.py

  people = Person.find(
      (Person.skills << desired_skill) &
      (Person.address.city == city)
  ).all()

此处的操作<<符用来表示“在……中”或者“包含”。

让我们找到谢菲尔德的所有吉他手：

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，我们调用deletePerson 类上的类方法来执行此操作：

  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-cli，你可以使用 Redis 命令检查该人现在是否设置了 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
给此页面评分
返回顶部 ↑