使用VSCode开发FastAPI指南（二）

在使用VSCode开发FastAPI指南（一）中，演示了如何在VS Code中使用FastAPI开发简单应用，包括环境设置和代码实现。首先创建Python虚拟环境并安装fastapi、redis等依赖项。然后编写基础API路由，使用Pydantic定义数据模型，实现添加/查询商品功能。教程详细介绍了调试配置、端口设置以及通过/docs界面测试API的方法。本文继续介绍数据存储相关的内容。

6设置数据存储

此时，我们已经拥有具有基本功能的应用程序的工作版本。本节将指导设置持久性数据存储。
到目前为止，我们将数据存储在字典中，这并不理想，因为当应用程序重新启动时，所有数据都会丢失。
为了持久保存数据，我们将使用 Redis，这是一个开源的内存数据结构存储。由于其速度和多功能性，Redis 通常用作各种应用程序中的数据存储系统，包括 Web 应用程序、实时分析系统、缓存层等。
如果是在Linux、macOS中开发，可以根据该系统的Redis指南来安装Redis。如果在Windwos中开发，可以参照以下内容。

6.1在 Windows 中设置 Docker 容器

VS Code Dev Containers 扩展提供了一种简化的方法，可将项目、其依赖项和所有必要的工具整合到一个整洁的容器中，从而创建一个功能齐全的开发环境。该扩展允许你在 VS Code 的容器内（或挂载到容器中）打开项目，在那里你将拥有其完整的功能集。

对于以下步骤，请确保计算机上安装了：

• 适用于 Windows 的 Docker
• Dev Containers 扩展

6.2创建 Dev 容器配置

打开命令面板并运行 Dev Containers：添加 Dev Container 配置文件…。
选择 Python 3：

选择默认版本。
选择 Redis Server 作为要安装的附加功能，按 OK，然后选择 Keep Defaults。
我们可以选择安装要包含在容器中的功能。在本教程中，我们将安装 Redis Server，这是一个社区提供的功能，用于为 Redis 安装和添加适当的开发容器设置。

在 Dev Containers 配置文件列表中选择 Redis Server 选项
这将在你的工作区.devcontainer中创建一个包含文件devcontainer.json的文件夹。让我们对此文件进行一些编辑，以便容器设置包括安装我们需要的 VS Code 扩展以及项目依赖项等步骤。
打开文件devcontainer.json
在条目"features" : { … }后添加 “，” ，以便我们可以向文件添加更多设置。
接下来，我们将向devcontainer.json文件中的postCreateCommand属性添加必要的依赖项安装命令，以便我们的应用程序在设置容器后可以运行。
找到下面的内容并从该行中删除注释 （//），以便在创建容器后可以安装依赖项：

"postCreateCommand": "pip3 install --user -r requirements.txt",

你可以在 Development Containers 规范中了解postCreateCommand和更多生命周期脚本。
现在，我们将使用customizations属性添加要在容器中安装的 VS Code 扩展。
将以下设置添加到devcontainer.json ：

// Use 'postCreateCommand' to run commands after the container is created.
"postCreateCommand": "pip3 install --user -r requirements.txt",// Configure tool-specific properties.
"customizations": {"vscode": {"extensions": ["ms-python.python", //Python extension ID"ms-python.vscode-pylance" //Pylance extension ID]}
}

保存文件。
从右下角显示的通知中选择 Reopen in Container，或从 Command Palette 中运行 Dev Containers： Reopen in Container 命令。

完成后，你将拥有一个完全配置的基于 Linux 的工作区，并安装了 Python 3 和 Redis Server。
设置容器后，你会注意到 VS Code 左下角有一个指示器：

注： 打开 Extensions （扩展） 视图 （Ctrl+Shift+X） 并进行搜索，仔细检查 Python 和 Pylance 扩展是否已成功安装在容器中。如果没有，你可以通过运行 Install in Dev Container 来安装它们。

所选的 Python 解释器信息位于右下角的 Status （状态） 栏上，与devcontainer.json中指定的版本匹配：

注意：如果你在状态栏上找不到 Python 解释器信息，可以单击 Python 解释器指示器（或从命令面板中运行 Python： Select Interpreter 命令），然后在容器中手动选择 Python 解释器。
我们现在已准备好进入下一部分，我们将在其中替换数据存储。

7替换数据库

我们有一个存储杂货清单项目的字典，但我们想用 Redis 数据库替换它。在本教程中，我们将使用 Redis 哈希来存储我们的数据，这是一种可以存储多个键值对的数据结构。
与传统数据库不同，你可以在不知道项目 ID 的情况下检索项目，你需要知道 Redis 哈希键才能从中检索值。在本教程中，我们将创建一个名为item_name_to_id的哈希来根据名称检索项目，并将它们映射到它们的 ID。此外，我们将创建其他哈希值以按 ID 检索项目，并将它们映射到它们的名称和数量。每个项目哈希都已命名为item_id:{item_id}，并且有两个字段：item_name和quantity .
首先，让我们先将字典替换为连接到 Redis 服务器的 Redis 客户端对象。
(1)在文件main.py中，将文件开头的grocery_list: dict[int, ItemPayload] = {} 替换为以下行：

redis_client = redis.StrictRedis(host='0.0.0.0', port=6379, db=0, decode_responses=True)

Pylance 将显示一条错误消息，因为 Redis 尚未导入。
(2)将光标放在编辑器中的 “redis” 上，然后单击显示的灯泡（或 Ctrl+.）。然后选择 Add ‘import redis’。

提示： 你可以通过在 Settings （设置） 编辑器 （Ctrl+，） 中查找 Auto Import Completions （自动导入完成） 设置并启用它，将 Pylance 设置为自动添加导入。
现在，我们有一个 Redis 客户端对象，它连接到在本地主机 （host=“0.0.0.0”） 上运行并侦听端口 6379 （port=6379） 的 Redis 服务器。参数db指定要使用的 Redis 数据库。Redis 支持多个数据库，在此代码中，我们将使用数据库 0，这是默认数据库。我们还将响应decode_responses=True解码为字符串（而不是字节）。
让我们在第一条路由add_item中再做一些替换。我们可以直接从 Redis 哈希中获取该信息，而不是查看字典中的所有键来查找已提供的项名称。
我们假设item_name_to_id哈希值已经存在，并将项目名称映射到它们的 ID。然后，我们可以通过从 Redis 调用hget方法来获取我们在请求中收到的项目名称的 ID，如果请求的名称已经存在于哈希中，它将返回项目 ID，如果不存在将返回None。

(3)删除内容如下的行：

items_ids = {item.item_name: item.item_id if item.item_id is not None else 0 for item in grocery_list.values()}

并将其替换为：

  item_id = redis_client.hget("item_name_to_id", item_name)

请注意，对此更改Pylance 会显示一个问题。这是因为hget方法返回str 或None （如果项目不存在）。但是，我们尚未替换的代码下方期待item_id的类型为int .让我们通过重命名item_id标识来解决此警告。
(4)重命名item_id为item_id_str
(5)如果启用了 inlay 提示，则 Pylance 应在 item_id_str旁边显示变量类型提示。你可以选择双击以接受它：

(6)如果项不存在，则item_id_str为None。所以现在我们可以删除包含以下内容的行：

if item_name in items_ids.keys():

并将其替换为：

if item_id_str is not None:

现在，我们有了商品 ID 作为字符串，我们需要将其转换为int并更新商品的数量。目前，我们的 Redis 哈希仅将项目名称映射到它们的 ID。为了将项目 ID 映射到它们的名称和数量，我们将为每个项目创建一个单独的 Redis 哈希，"item_id:{item_id}"用作我们的哈希名称，以便更轻松地按 ID 进行检索。我们还将为每个哈希添加item_name和quantity字段。

(7)删除if块中的代码：

item_id: int = items_ids[item_name]
grocery_list[item_id].quantity += quantity

并添加以下内容，将item_id 转换为int ，然后通过从 Redis 调用hincrby方法来增加项目的数量。此方法将"quantity"字段的值递增请求 （quantity） 中的给定量：

item_id = int(item_id_str)
redis_client.hincrby(f"item_id:{item_id}", "quantity", quantity)

我们现在只需要替换当item_id_str 为None 时的代码。在这种情况下，我们生成一个新的item_id ，为项目创建一个新的 Redis 哈希，然后添加提供的项目名称和数量。
要生成新的item_id ，让我们使用 Redis 中的incr方法，传递一个名为"item_ids"的新哈希，此哈希用于存储最后生成的 ID，因此我们可以在每次创建新项目时递增它，确保它们都具有唯一 ID。

(8)删除如下内容的行：

item_id: int = max(grocery_list.keys()) + 1 if grocery_list else 0

并添加以下内容：

item_id: int = redis_client.incr("item_ids")

首次使用item_ids键运行incr调用时， Redis 会创建键并将其映射到值1 。然后，每次后续运行时，它都会将存储的值增加 1。
现在，我们将使用hset方法并通过为字段 （item_id、item_name 和 quantity） 和值 （项目新创建的 ID 及其提供的名称和数量） 提供映射，将项目添加到 Redis 哈希中。

(9)删除如下内容的行：

grocery_list[item_id] = ItemPayload(item_id=item_id, item_name=item_name, quantity=quantity)

并将其替换为以下内容：

redis_client.hset(f"item_id:{item_id}",mapping={"item_id": item_id,"item_name": item_name,"quantity": quantity,})

现在我们只需要通过设置我们在开头引用的哈希值，将新创建的 ID 映射到项目名称item_name_to_id.
(10)将这一行添加到路由的末尾，在else块内：

redis_client.hset("item_name_to_id", item_name, item_id)

(11)删除内容如下的行：

return {"item": grocery_list[item_id]}

并将其替换为：

return {"item": ItemPayload(item_id=item_id, item_name=item_name, quantity=quantity)}

(12)如果你愿意，你可以尝试对其他路线进行类似的替换。否则，你只需将文件的全部内容替换为以下内容：

import redis
from fastapi import FastAPI, HTTPExceptionfrom models import ItemPayloadapp = FastAPI()redis_client = redis.StrictRedis(host="0.0.0.0", port=6379, db=0, decode_responses=True)# Route to add an item
@app.post("/items/{item_name}/{quantity}")
def add_item(item_name: str, quantity: int) -> dict[str, ItemPayload]:if quantity <= 0:raise HTTPException(status_code=400, detail="Quantity must be greater than 0.")# Check if item already existsitem_id_str: str | None = redis_client.hget("item_name_to_id", item_name)if item_id_str is not None:item_id = int(item_id_str)redis_client.hincrby(f"item_id:{item_id}", "quantity", quantity)else:# Generate an ID for the itemitem_id: int = redis_client.incr("item_ids")redis_client.hset(f"item_id:{item_id}",mapping={"item_id": item_id,"item_name": item_name,"quantity": quantity,},)# Create a set so we can search by name tooredis_client.hset("item_name_to_id", item_name, item_id)return {"item": ItemPayload(item_id=item_id, item_name=item_name, quantity=quantity)}# Route to list a specific item by ID but using Redis
@app.get("/items/{item_id}")
def list_item(item_id: int) -> dict[str, dict[str, str]]:if not redis_client.hexists(f"item_id:{item_id}", "item_id"):raise HTTPException(status_code=404, detail="Item not found.")else:return {"item": redis_client.hgetall(f"item_id:{item_id}")}@app.get("/items")
def list_items() -> dict[str, list[ItemPayload]]:items: list[ItemPayload] = []stored_items: dict[str, str] = redis_client.hgetall("item_name_to_id")for name, id_str in stored_items.items():item_id: int = int(id_str)item_name_str: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")if item_name_str is not None:item_name: str = item_name_strelse:continue  # skip this item if it has no nameitem_quantity_str: str | None = redis_client.hget(f"item_id:{item_id}", "quantity")if item_quantity_str is not None:item_quantity: int = int(item_quantity_str)else:item_quantity = 0items.append(ItemPayload(item_id=item_id, item_name=item_name, quantity=item_quantity))return {"items": items}# Route to delete a specific item by ID but using Redis
@app.delete("/items/{item_id}")
def delete_item(item_id: int) -> dict[str, str]:if not redis_client.hexists(f"item_id:{item_id}", "item_id"):raise HTTPException(status_code=404, detail="Item not found.")else:item_name: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")redis_client.hdel("item_name_to_id", f"{item_name}")redis_client.delete(f"item_id:{item_id}")return {"result": "Item deleted."}# Route to remove some quantity of a specific item by ID but using Redis
@app.delete("/items/{item_id}/{quantity}")
def remove_quantity(item_id: int, quantity: int) -> dict[str, str]:if not redis_client.hexists(f"item_id:{item_id}", "item_id"):raise HTTPException(status_code=404, detail="Item not found.")item_quantity: str | None = redis_client.hget(f"item_id:{item_id}", "quantity")# if quantity to be removed is higher or equal to item's quantity, delete the itemif item_quantity is None:existing_quantity: int = 0else:existing_quantity: int = int(item_quantity)if existing_quantity <= quantity:item_name: str | None = redis_client.hget(f"item_id:{item_id}", "item_name")redis_client.hdel("item_name_to_id", f"{item_name}")redis_client.delete(f"item_id:{item_id}")return {"result": "Item deleted."}else:redis_client.hincrby(f"item_id:{item_id}", "quantity", -quantity)return {"result": f"{quantity} items removed."}

(13)重新运行调试器，通过与/docs路由交互来测试此应用程序。完成后，你可以停止调试器。

现在，你有一个正在运行的 FastAPI 应用程序，其中包含用于在杂货清单中添加、列出和删除商品的路由，并且数据保存在 Redis 数据库中。

8设置数据库删除

现在，Redis 保留了数据，你可能需要创建一个脚本来删除所有测试数据。为此，使用以下内容创建一个新文件flushdb.py：

import redisredis_client = redis.StrictRedis(host='0.0.0.0', port=6379, db=0, decode_responses=True)
redis_client.flushdb()

然后，当你想要重置数据库时，可以在 VS Code 中打开文件flushdb.py，然后选择编辑器右上角的“运行”按钮，或从命令面板运行 Python：在终端中运行 Python 文件命令。

请注意，应谨慎执行此作，因为它会删除当前数据库中的所有键，如果在生产环境中执行此作，可能会导致数据丢失。