简介

在我的上一篇文章中,我写过使用 Redis 作为消息代理来存储任务队列传递的消息。除了 Redis 用作缓存存储或消息代理之外,Redis 还可以用作数据库,以键值对的形式存储数据。

在本文中,我们将通过构建博客服务来探索 RedisJson 和 RedisSearch 的功能,以使用 Redis-OM 和 python 最快的 Web 框架 - FastAPI 创建、检索、更新和执行全文搜索。

目标

在本教程结束时,您将能够:

• 在Redis上创建Redis数据库

• 使用 Redis-OM 创建数据库模型

• 使用 Python FastAPI 开发 Restful API

• 与Redis交互创建、检索和搜索数据

环境设置

在您的桌面上创建一个新文件夹并将其命名为“blog-fast-API”

$ mkdir blog-fast-API

$ cd blog-fastAPI

创建一个虚拟环境并激活它

对于 Windows 用户

$ python3  -m venv env 
$ cd env 
$ cd Scripts 
$ activate

对于 macOS 用户

$ python3 -m venv env 
$ source env/bin/activate

在您的根目录中,创建一个名为 blog 的文件夹,并在 blog 下创建两个名为models.py和main.py的新文件。另外,创建一个 .gitignore 文件并添加env以在推送到 Github 时忽略您的环境变量。

您的目录应如下所示:

通过运行安装依赖项

$ pip install fastapi
$ pip install redis-om
$ pip install "uvicorn[standard]"

FastAPI是一个现代、高性能的 Web 框架,用于在 Python 中开发 RESTful API。它快速、简单并自动生成 swagger API 文档。

运行pip freeze > requirements.txt以确保所有依赖项都在您的环境中。

在blog/main.py文件中,让我们创建一个根路由来运行服务器。

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello world"}

进入博客目录并运行服务器

$ cd blog
$ uvicorn main:app --reload

在浏览器中导航到 http://127.0.0.1:8000,您应该会看到:

{"message":"Hello world"}

FastAPI 自动为 API 生成 swagger 文档。您可以通过在浏览器中导航到127.0.0.1:8000/docs来查看您的 API 文档。

数据库设置

我们将在 Redis 云上创建一个新数据库。

在Redis上创建一个帐户并登录。

导航到仪表板以创建新订阅。您可以从免费套餐开始,稍后根据您的应用使用情况进行升级。

创建一个新数据库并激活它,我将调用我的“博客”。

在您的blog/models.py文件中,导入get_redis_connection并连接到您的 Redis 数据库。

from redis_om import get_redis_connection

redis = get_redis_connection(
    host="redis-14531.c13.us-east-1-3.ec2.cloud.redislabs.com",
    port=14531,
    password="TbuxEUkupHczh3G5SVYFr2tF1SPmILDk",
    decode_responses=True,
)

host是从仪表板上的数据库配置中获取的公共端点上的基本 URL,它在.com处停止。

port是公共端点上的最后五位数字

password是仪表板上数据库配置中的默认用户密码。

创建模型

我们将为我们的应用程序创建两个模型,Author模型和Blog模型。 Author 模型将嵌套到 Blog 模型中。

在您的“博客/models.py”中导入EmbeddedJsonModel和JsonModel并创建您的模型

import datetime

from redis_om import get_redis_connection, EmbeddedJsonModel, JsonModel, Field, Migrator

redis = get_redis_connection(
    host="redis-14531.c13.us-east-1-3.ec2.cloud.redislabs.com",
    port=14531,
    password="TbuxEUkupHczh3G5SVYFr2tF1SPmILDk",
    decode_responses=True,
)


class Author(EmbeddedJsonModel):
    first_name: str = Field(index=True, full_text_search=True)
    last_name: str
    email: str
    bio: str
    date_joined: datetime.date = Field(default=datetime.datetime.now())

    class Meta:
        database = redis


class Blog(JsonModel):
    title: str = Field(index=True, full_text_search=True)
    content: str
    author: Author
    date_posted: datetime.date = Field(
        default=datetime.datetime.today().strftime("%Y-%m-%d")
    )

    class Meta:
        database = redis


Migrator().run()

在上面的代码中,我们定义了表示我的数据将如何存储在 Redis 数据库中的模式。

EmbeddedJsonModel类允许在Blog模型中嵌套Author模型。Field允许定义额外的功能。index=True告诉 Redis-OM 我们要索引 first_name 和 title 字段。full_text_search=True告诉 Redis-OM 我们要执行 full_text_search 查询。default以年/月/日格式设置默认日期。Migrator().run()运行迁移以设置 Redis OM 在我们开始查询之前将使用的索引。

Redis OM 的高级功能,如将 JSON 模型相互嵌套并执行丰富的查询表达式,依赖于两个 Redis 模块:RedisJson 和 RedisSearch。RedisJson允许我们使用 Redis 添加 JSON 数据类型,而RedisSearch允许我们使用 Redis 执行查询、索引和 full_text_search。

创建 CRUD API

我们将创建 API 来运行 CRUD 操作。

在main.py文件中,导入模型并开始创建 API

from fastapi import FastAPI
from models import Author, Blog

from redis_om.model import NotFoundError

app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello world"}


# create an author
@app.post("/authors")
async def create_author(body: Author):
    author = Author(
        first_name=body.first_name,
        last_name=body.last_name,
        email=body.email,
        bio=body.bio,
        date_joined=body.date_joined,
    )

    author.save()

    return author


# create a blog
@app.post("/blogs")
async def create_blog(body: dict):
    author = Author.get(body["author_id"])
    blog = Blog(title=body["title"], content=body["content"], author=author)

    blog.save()

    return blog


# get a blog
@app.get("/blogs/{pk}")
async def get_blog(pk: str):
    try:
        blog = Blog.get(pk)
    except NotFoundError:
        return {"error": "no resource found"}, 404
    return blog


# update a blog
@app.put("/blogs/{pk}")
async def update_blog(pk: str, body: dict):
    blog = Blog.get(pk)

    blog.title = body["title"]
    blog.content = body["content"]

    blog.save()

    return blog


# delete a blog
@app.delete("/blogs/{pk}")
async def delete_blog(pk: str):
    Blog.delete(pk)
    return {"success": "blog deleted successfully"}

在上面的代码中,我们添加了创建作者、创建博客文章、获取博客文章、更新博客文章和删除博客文章的基本端点。

create a blog post 将 author_id 作为请求体来查询作者并在创建新博客帖子时将其嵌入。我们还通过导入 Redis OM 提供的NotFoundError类来处理未找到博客文章的错误。

测试 CRUD API

现在让我们测试我们在上面创建的 API。

在浏览器中转到127.0.0.1:8000/docs以查看由 FastAPI 生成的 Swagger 文档。

创建作者

在创建新博客时将作者 pk 保存在某处以供输入。

创建博客

从上面的响应中,作者的详细信息嵌套在博客响应中。这是 RedisJson 的特性之一。

获取博客

让我们传递之前创建的博客的 pk 并测试端点。

更新博客

让我们通过传递 pk 更新创建的博客

删除博客

让我们通过博客的pk来测试endpoint

创建搜索 API

在blog/main.py中添加如下代码

def format_results(data):
    response = []
    for dat in data:
        response.append(dat.dict())

    return {"results": response}


@app.post("/blogs/find")
async def blog_by_name(title: str):
    blogs = Blog.find(Blog.title % title).all()

    return format_results(blogs)


@app.post("/blogs/find/author")
async def blog_by_author(first_name: str):
    blogs = Blog.find(Blog.author.first_name == first_name).all()

    return format_results(blogs)

在上面的代码中,我们创建了两个 API 来在数据库中搜索博客文章。blog_by_nameAPI 搜索标题包含与搜索词匹配的全文搜索的博客文章。blog_by_authorAPI 通过作者的名字返回博客文章。我们还创建了一个函数来格式化这些 API 返回的响应。

现在,让我们测试 API

测试搜索 API

按名称查找

创建不同的博客文章,并在标题中使用相似的单词进行测试。

让我们输入一个博客通用的关键字,以过滤掉与该关键字匹配的博客。

按作者查找

让我们输入一个创建的作者的名字来过滤掉该作者的所有博客文章。

结论

在本教程中,您学习了如何使用 FastAPI 和 Redis OM 创建执行 crud 操作的 API。您已经通过存储 JSON 文档并检索它们来探索 RedisJson 的功能。您还使用了 RedisSearch 功能来执行查询和 full_text_search。

代码可以在这个存储库中找到。

这篇文章是与 Redis 合作的。

学到更多:

• 免费试用 Redis Cloud

• 观看此视频,了解 Redis Cloud 相对于其他 Redis 提供商的优势

• Redis Developer Hub - 关于 Redis 的工具、指南和教程

• RedisInsight 桌面 GUI