Visual Studio Code 中的 FastAPI 教程

FastAPI是一个现代高性能 Web 框架,用于使用 Python 构建 API。它旨在让您轻松快速、高效地构建 API,同时提供自动验证、序列化和 API 文档等功能,使其成为构建 Web 服务和微服务的热门选择。

在本 FastAPI 教程中,我们将使用 FastAPI 创建一个杂货清单应用程序。在本教程结束时,您将了解如何在 Visual Studio Code 终端、编辑器和调试器中使用 FastAPI。本教程不是 FastAPI 深入探讨。为此,您可以参考FastAPI官方文档

如果这是您第一次使用 Python,我们建议您从我们的Python 教程开始,以熟悉该语言和 VS Code 的 Python 支持。本教程更适合那些已经熟悉 Python 并想要学习如何在 VS Code 中使用 FastAPI 的人。

本 FastAPI 教程中已完成的代码项目可以在 GitHub 上找到:python-sample-vscode-fastapi-tutorial

如果您有任何问题,可以在Python 扩展 讨论问答上搜索答案或提问。

设置项目

您可以通过不同的方式为本教程设置项目。我们将介绍如何在本地计算机上的GitHub Codespaces和VS Code中进行设置。

GitHub 代码空间

您可以设置此项目在GitHub Codespaces中进行开发,您可以在代码空间中远程编码、调试和运行您的应用程序。代码空间提供托管在云中的完全配置的开发环境,无需本地设置。该环境包括项目的依赖项、工具和扩展,确保一致且可重复的开发体验。它通过提供实时编辑、集成​​版本控制以及轻松访问调试和测试工具来简化协作,同时保持项目的安全性和可靠性。

注意:所有 GitHub.com 帐户都有每月免费使用免费或专业计划中包含的 GitHub Codespaces 的配额。有关更多信息,请转到关于 GitHub Codespaces 的计费

要为本教程设置代码空间,请导航到该项目的 GitHub 存储库。该代码空间包含快速开始 FastAPI 开发所需的所有必要配置和依赖项。

对于本教程,选择基于字典的分支:

在 python-sample-vscode-fastapi-tutorial GitHub 存储库中选择的基于字典的分支

然后,选择Code > Codespaces > Create Codespace on <dictionarybased>分支为您的项目创建并打开代码空间。

完成后,您可以继续下面的替换数据库部分。

本地 VS Code 中

要在VS Code中成功完成本教程,您首先需要设置 Python 开发环境。具体来说,本教程需要:

在本节中,我们将创建一个文件夹以在 VS Code 中作为工作区打开,设置 Python 虚拟环境,并安装项目的依赖项。

  1. 在您的文件系统中,为本教程创建一个项目文件夹,例如groceries-plugin.

  2. 在 VS Code 中打开这个新文件夹(文件>打开文件夹...)。

  3. 当出现“工作空间信任”提示时,选择“是,我信任作者”以允许工作空间访问必要的资源和扩展。您可以在文档中了解有关 Workspace Trust 的更多信息。

现在,让我们创建一个requirements.txt文件,列出我们希望为应用程序安装的依赖项。该requirements.txt文件是Python开发中的常见做法,用于指定项目所依赖的库及其版本。该文件有助于确保从事该项目的任何人都可以重新创建类似的开发环境,使其成为维护一致性的便捷组件。

我们将安装 FastAPI 来创建应用程序,安装uvicorn作为服务器,安装Redistype-redis处理数据存储并与 Redis 数据库交互。

  1. 在 VS Code 中创建一个新文件(“文件” > “新建文本文件”⌘N(Windows、Linux Ctrl+N)。

  2. 添加以下内容:

    fastapi
    redis
    types-redis
    uvicorn
    
  3. 保存文件(⌘S(Windows、Linux Ctrl+S)并命名requirements.txt

  4. 通过打开命令面板 ( ⇧⌘P (Windows、Linux Ctrl+Shift+P ) ) 并运行Python: Create Environment命令来创建虚拟环境。

    注意:此步骤可能需要几分钟才能完成。

  5. 当询问环境类型时,选择Venv

    下拉菜单中包含“Venv”或“Conda”作为可使用 Python 创建的环境选项:创建环境命令

  6. 然后选择您计算机上可用的最新版本的 Python:

    可用于创建虚拟环境的可用全局环境列表

  7. 从下拉列表中选择requirements.txt文件,以便自动安装依赖项,然后选择“确定”

    选中复选框以安装requirements.txt 文件中的依赖项

将创建虚拟环境,自动安装依赖项,并为您的工作区选择供 Python 扩展使用的环境。您可以通过检查 VS Code 的右下角来确认它已被选中:

状态栏中的环境

注意:如果在状态栏上没有找到新创建的环境信息,您可以单击Python解释器指示器(或从命令面板运行Python:选择解释器命令)并手动选择虚拟环境。

开始编码

让我们创建应用程序!

  1. 使用“文件” > “新建文件...”创建一个新的 Python 文件,然后选择“Python 文件”

  2. 将其另存为main.py( ⇧⌘S (Windows, Linux Ctrl+Shift+S ) ) )在groceries-plugin文件夹中。

  3. 添加以下代码main.py并保存文件:

    from fastapi import FastAPI
    
    app = FastAPI()
    
    @app.get("/")
    def root():
        return {"message": "Hello World"}
    
  4. 通过启动调试器 ( F5 ) 来运行代码。

  5. 从下拉菜单中,从列表中选择FastAPI配置选项:

    带有调试器配置选项的下拉菜单,其中突出显示了 FastAPI

    这会调用 uvicorn 通过调试器启动应用程序服务器,并允许您单步执行源代码以检查其行为。您应该在终端中看到类似以下内容:

    Uvicorn 服务器运行消息显示在终端中,并带有用于访问应用程序的 URL

  6. 按住 Ctrl 键并单击http://127.0.0.1:8000/终端中的 URL 以打开该地址的默认浏览器:

    浏览器中显示的 Hello World 消息

    恭喜!您的 FastAPI 应用程序已启动并正在运行!

  7. 使用调试工具栏中的“停止”按钮或通过⇧F5(Windows、Linux Shift+F5停止调试器。

为杂货清单项目创建模型

现在我们已经有了 FastAPI 应用程序,我们可以使用Pydantic来定义我们的购物清单项目,Pydantic 是一个与 FastAPI 无缝集成的数据验证和解析库。Pydantic 允许您使用带有类型提示的 Python 类定义数据模型,以自动验证和解析 API 请求中的传入数据(称为“有效负载”)。

让我们为杂货清单项目创建一个模型。我们将使用该ItemPayload模型来定义要添加到购物清单中的商品的数据结构。该模型将具有三个字段:item_iditem_namequantity

  1. 使用File > New File...创建一个新的 Python 文件,然后选择Python File

  2. 将以下行添加到文件中,然后将其保存在groceries-plugin文件夹中models.py⇧⌘S(Windows、Linux Ctrl+Shift+S):

    from typing import Optional
    from pydantic import BaseModel
    
    class ItemPayload(BaseModel):
        item_id: Optional[int]
        item_name: str
        quantity: int
    

Pylance是 VS Code 中 Python 的默认语言服务器,支持类型提示功能,这有助于使用 Pydantic 模型和 FastAPI。这是因为 Pylance 构建在Pyright之上,Pyright 是 Python 的静态类型检查器,可以检测代码中的类型错误,以防止错误并提高代码质量。

下面的三个步骤是可选的,但鉴于 FastAPI 广泛使用类型提示来提高代码可读性和验证,我们可以利用 Pylance 的类型检查功能来尽早捕获错误:

  1. 打开设置编辑器 ( ⌘, (Windows, Linux Ctrl+, ))。

  2. 搜索“python 类型检查模式”并将其设置basic为基本类型检查。Pylance 现在将显示诊断和警告以捕获简单的与类型相关的错误。或者,您可以将其设置为strict强制执行更高级的类型检查规则

    设置编辑器中的 Python 分析类型检查模式选项(关闭、基本和严格)

  3. 接下来,搜索“Python嵌入类型提示”,并启用变量类型函数返回类型的嵌入提示:

    在设置编辑器中启用两个 Python 分析类型提示设置:函数返回类型和变量类型

创建路线

现在我们需要一个地方来存储杂货清单项目。为了简单起见,我们从一个空字典开始。

  1. 首先,让我们导入示例所需的所有包。打开main.py文件并将第一个导入行替换为以下行:

    from fastapi import FastAPI, HTTPException
    
    from models import ItemPayload
    
  2. 现在在下面添加以下行app = FastAPI()

    grocery_list: dict[int, ItemPayload] = {}
    

    这将创建一个新的空字典,用于接收类型的键int(作为项目 ID)和该ItemPayload类型的值。

我们现在将在 FastAPI 应用程序中定义路由。在 Web 应用程序的上下文中,路由就像将特定 URL 映射到处理它们的代码的路径。这些路由充当我们应用程序中不同功能的入口点。当客户端(例如 Web 浏览器或其他程序)使用特定 URL 向我们的应用程序发送请求时,FastAPI 会根据 URL 将该请求路由到适当的函数(也称为路由处理程序或视图函数),并且该函数处理请求并生成响应。

让我们继续定义添加和检索单个商品的路由,以及返回购物清单中的所有商品。

  1. 在文件末尾添加以下路由main.py

    # Route to add a item
    @app.post("/items/{item_name}/{quantity}")
    def add_item(item_name: str, quantity: int):
        if quantity <= 0:
            raise HTTPException(status_code=400, detail="Quantity must be greater than 0.")
        # if item already exists, we'll just add the quantity.
        # get all item names
        items_ids = {item.item_name: item.item_id if item.item_id is not None else 0 for item in grocery_list.values()}
        if item_name in items_ids.keys():
            # get index of item_name in item_ids, which is the item_id
            item_id = items_ids[item_name]
            grocery_list[item_id].quantity += quantity
    # otherwise, create a new item
        else:
            # generate an ID for the item based on the highest ID in the grocery_list
            item_id = max(grocery_list.keys()) + 1 if grocery_list else 0
            grocery_list[item_id] = ItemPayload(
                item_id=item_id, item_name=item_name, quantity=quantity
            )
    
        return {"item": grocery_list[item_id]}
    

    如果您在上一节中启用了类型提示,您可能会注意到 Pylance 添加了函数返回类型以及 和 的类型的嵌入item_ids提示item_id。您可以选择双击每个建议将其插入代码中:

    Pylance 在整个示例代码中显示嵌入函数返回和变量类型提示

现在让我们检查一下这条路线是否按预期工作。最快的方法是使用 VS Code 的调试器和 FastAPI 的/docs端点,它提供有关所有可用 API 路由的信息,并允许您与 API 交互以探索其参数和响应。该文档是根据 FastAPI 应用程序中定义的元数据和类型提示动态生成的。

  1. if quantity <= 0通过单击行号的左边距(或F9)在语句旁边添加断点。调试器将在执行该行之前停止,因此您可以逐行检查代码。

    add_item 函数中第一行旁边设置的断点

  2. 启动调试器 ( F5 ),然后导航到http://127.0.0.1:8000/docs浏览器中。

    应用程序中应该有一个 Swagger 接口,其中包含两个可用端点:/items和 root ( /)。

    Swagger UI 显示两个端点:/items 和 /

  3. 选择路线旁边的向下箭头/items将其展开,然后选择右侧显示的“试用”按钮。

    Swagger UI 中 /items 路由旁边显示的“尝试一下”按钮

  4. 通过将字符串传递到item_name字段并将数字传递到 来添加杂货清单项目quantity。例如,您可以提供 apple 作为 ,提供item_name2 作为quantity

  5. 选择执行

    /items 路由下方显示的执行按钮

  6. 再次打开 VS Code,您会发现调试器已停止在您之前设置的断点处。

    调试器在 add_item 函数中设置的断点处停止

    在左侧,此时定义的所有局部和全局变量都显示在“变量”窗口的“运行”和“调试”视图下。在我们的示例中,item_name被设置为“apple”,并quantity在局部变量视图下设置为 2,以及grocery_list在全局变量视图下设置为空字典。

    “运行”和“调试”视图中显示“变量”窗口,其中突出显示了 item 和grocery_list 变量

现在让我们使用 VS Code 的调试控制台来进行一些探索。

  1. 选择该quantity <= 0语句,右键单击编辑器并选择Evaluate in Debug Console

    右键单击代码行时,在上下文菜单中显示“在调试控制台中评估”选项

    这将打开调试控制台并运行选定的表达式。正如我们示例中所预期的那样,表达式的计算结果为False

调试控制台是一个强大的工具,可以快速测试表达式并更好地了解断点时代码的状态。您还可以使用它来运行任意代码,例如调用函数或打印变量。您可以在Python 教程中了解有关 VS Code 中的 Python 调试的更多信息。

现在,您可以通过选择“调试”视图工具栏中的“继续”或按F5来继续执行代码。

最后,让我们为应用程序添加剩余的路线,以便我们可以列出所有项目或特定项目,以及将它们从我们的购物清单中删除。您可以让调试器保持运行状态,因为当您保存下一步中所做的更改时,它会自动重新加载应用程序。

  1. 将其中的内容替换main.py为以下代码:

    from fastapi import FastAPI, HTTPException
    
    from models import ItemPayload
    
    app = FastAPI()
    
    grocery_list: dict[int, ItemPayload] = {}
    
    # 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.")
        # if item already exists, we'll just add the quantity.
        # get all item names
        items_ids: dict[str, int] = {
            item.item_name: item.item_id if item.item_id is not None else 0
            for item in grocery_list.values()
        }
        if item_name in items_ids.keys():
            # get index of item_name in item_ids, which is the item_id
            item_id: int = items_ids[item_name]
            grocery_list[item_id].quantity += quantity
        # otherwise, create a new item
        else:
            # generate an ID for the item based on the highest ID in the grocery_list
            item_id: int = max(grocery_list.keys()) + 1 if grocery_list else 0
            grocery_list[item_id] = ItemPayload(
                item_id=item_id, item_name=item_name, quantity=quantity
            )
    
        return {"item": grocery_list[item_id]}
    
    
    # Route to list a specific item by ID
    @app.get("/items/{item_id}")
    def list_item(item_id: int) -> dict[str, ItemPayload]:
        if item_id not in grocery_list:
            raise HTTPException(status_code=404, detail="Item not found.")
        return {"item": grocery_list[item_id]}
    
    
    # Route to list all items
    @app.get("/items")
    def list_items() -> dict[str, dict[int, ItemPayload]]:
        return {"items": grocery_list}
    
    
    # Route to delete a specific item by ID
    @app.delete("/items/{item_id}")
    def delete_item(item_id: int) -> dict[str, str]:
        if item_id not in grocery_list:
            raise HTTPException(status_code=404, detail="Item not found.")
        del grocery_list[item_id]
        return {"result": "Item deleted."}
    
    
    # Route to remove some quantity of a specific item by ID
    @app.delete("/items/{item_id}/{quantity}")
    def remove_quantity(item_id: int, quantity: int) -> dict[str, str]:
        if item_id not in grocery_list:
            raise HTTPException(status_code=404, detail="Item not found.")
        # if quantity to be removed is higher or equal to item's quantity, delete the item
        if grocery_list[item_id].quantity <= quantity:
            del grocery_list[item_id]
            return {"result": "Item deleted."}
        else:
            grocery_list[item_id].quantity -= quantity
        return {"result": f"{quantity} items removed."}
    
    
  2. 保存文件 ( ⌘S (Windows、Linux Ctrl+S ) )。应用程序应该自动重新加载。

现在,您可以/docs再次打开页面并使用调试器和调试控制台测试新路由,以更好地了解代码执行情况。完成后,您可以停止调试器 ( ⇧F5 (Windows、Linux Shift+F5 ) )。您还可以通过单击删除我们在步骤 4 中添加的断点。

恭喜!现在,您已经有了一个可以运行的 FastAPI 应用程序,其中包含用于添加、列出和删除购物清单中的商品的路由。

设置数据存储

此时,您已经拥有了具有基本功能的应用程序的工作版本。本部分将指导您完成设置数据存储以实现持久性,但如果您对已经学到的内容感到满意,则可以选择跳过它。

到目前为止,我们将数据存储在字典中,这并不理想,因为应用程序重新启动时所有数据都会丢失。

为了持久保存数据,我们将使用Redis,它是一个开源内存数据结构存储。由于其速度和多功能性,Redis 通常用作各种应用程序中的数据存储系统,包括 Web 应用程序、实时分析系统、缓存层、本教程等。

如果您已经使用我们现有的模板使用GitHub Codespaces,则可以直接跳到替换数据库部分。

如果您使用的是 Windows,则可以通过设置Docker 容器GitHub Codespace或通过WSL(适用于 Linux 的 Windows 子系统)来使用 Redis。在本教程中,我们将使用 Docker 容器,但您可以参考上面的部分,了解如何设置 GitHub Codespace 的说明。

否则,如果您使用的是 Linux 或 macOS 计算机,则可以按照其网站上的说明安装 Redis ,然后跳到替换数据库部分。

在 Windows 上设置 Docker 容器

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

对于以下步骤,请确保您的计算机上安装了以下要求:

要求

创建 Dev 容器配置

  1. 打开命令面板并运行 Dev Containers :添加 Dev Containers 配置文件...。

  2. 选择Python 3

    在 Dev Containers 配置文件列表中选择 Python 3 选项

  3. 选择默认版本。

我们可以选择安装要包含在容器中的功能。在本教程中,我们将安装Redis Server,这是一个社区贡献的功能,可为 Redis 安装并添加正确的 Dev Containers 设置。

  1. 选择Redis Server作为要安装的附加功能,按“确定”,然后选择“保留默认值”

    在 Dev Containers 配置文件列表中选择 Redis 服务器选项

这将.devcontainer在您的工作区中创建一个包含devcontainer.json文件的文件夹。让我们对此文件进行一些编辑,以便容器设置包括安装我们需要的 VS Code 扩展以及项目依赖项等步骤。

  1. 打开devcontainer.json文件。

  2. 在条目后面添加一个“,” "features" : { ... },以便我们可以向文件添加更多设置。

接下来,我们将向postCreateCommand文件中的属性添加必要的依赖项安装命令devcontainer.json,以便在容器设置完毕后我们的应用程序就可以运行了。

  1. 找到下面的内容并删除该行中的注释 ( //),以便在创建容器后可以安装依赖项:

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

    您可以在 Dev Containers 规范中了解postCreateCommand更多生命周期脚本。

    现在我们将使用该customizations属性添加我们想要安装在容器中的 VS Code 扩展。

  2. 添加以下设置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
                ]
            }
        }
    
  3. 保存文件。

  4. 从右下角显示的通知中选择“在容器中重新打开”,或从命令面板运行“ Dev Containers :在容器中重新打开”命令。

    注意:构建容器可能需要几分钟的时间,具体取决于互联网速度和机器性能。

    您可以在 Dev Containers 文档中了解有关 Dev Containers 配置的更多信息。

完成后,您将拥有一个完全配置的基于 Linux 的工作区,其中安装了 Python 3 和 Redis 服务器。

设置容器后,您会注意到 VS Code 左下角有一个指示器:

VS Code 左下角显示 Dev Containers 指示器

注意:通过打开扩展视图 ( ⇧⌘X (Windows、Linux Ctrl+Shift+X ) ) 并搜索,仔细检查Python 和 Pylance 扩展是否已成功安装在容器中。如果没有,您可以通过运行Install in Dev Container来安装它们。

右下角状态栏上提供了所选的Python解释器信息,与文件中指定的版本相匹配devcontainer.json

Python解释器选择

注意:如果在状态栏上没有找到Python解释器信息,您可以单击Python解释器指示器(或从命令面板运行Python:选择解释器命令)并手动选择容器中的Python解释器。

我们现在准备好进入下一部分,我们将替换数据存储。

更换数据库

我们有一个存储杂货清单项目的字典,但我们想用 Redis 数据库替换它。在本教程中,我们将使用 Redis 哈希来存储数据,这是一种可以存储多个键值对的数据结构。

与传统数据库不同,您无需知道项目 ID 即可检索项目,而您需要知道 Redis 哈希键才能从中检索值。在本教程中,我们将创建一个名为的哈希,用于按item_name_to_id名称检索项目,并将它们映射到其 ID。此外,我们将创建其他哈希来按 ID 检索项目,将它们映射到名称和数量。每个项目哈希均已命名item_id:{item_id}并具有两个字段:item_namequantity

首先,我们首先将字典替换为连接到 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 尚未导入。

  1. 将光标放在编辑器中的“redis”上,然后单击显示的灯泡(或⌘。(Windows、Linux Ctrl+.)。然后选择添加“导入redis”

    Redis 变量旁边显示灯泡,可以选择添加导入语句

    提示:您可以通过在“设置”编辑器中查找“自动导入完成”设置( ⌘,(Windows、Linux Ctrl+,)并启用它,将 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 中的方法来获取请求中接收到的项目名称的 ID hget,如果请求的名称已存在于哈希中,则该方法将返回项目 ID,或者None不存在。

  1. 删除包含以下内容的行:

    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符号来解决此警告。

  1. 重命名item_iditem_id_str.

  2. 如果您启用了嵌入提示,Pylance 应该在 旁边显示变量类型提示item_id_str。您可以选择双击接受它:

    item_id_str 变量旁边显示的变量类型提示

  3. 如果该项不存在,item_id_str则为None。现在我们可以删除包含以下内容的行:

    if item_name in items_ids.keys():
    

    并将其替换为:

    if item_id_str is not None:
    

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

  1. 删除块内的代码if

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

    并添加以下内容,将 转换item_idint,然后通过调用hincrbyRedis 中的方法来增加项目的数量。此方法将字段的值增加"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,让我们使用incrRedis 中的方法,传递一个名为 的新哈希"item_ids"。该散列用于存储最后生成的 ID,因此我们可以在每次创建新项目时递增它,确保它们都具有唯一的 ID。

  1. 删除包含以下内容的行:

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

    并添加以下内容:

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

    incr当第一次使用 key 运行此调用时item_ids,Redis 创建 key 并将其映射到 value 1。然后,每次运行时,它都会将存储的值增加 1。

现在,我们将使用该方法并通过提供字段( 、和)以及值(该项目新创建的 ID 及其提供的名称和数量)的hset映射,将项目添加到 Redis 哈希中。item_iditem_namequantity

  1. 删除包含以下内容的行:

    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

  1. 将此行添加到块内路线的末尾else

    redis_client.hset("item_name_to_id", item_name, item_id)
    
  2. 删除包含以下内容的行:

    return {"item": grocery_list[item_id]}
    

    并将其替换为:

    return {"item": ItemPayload(item_id=item_id, item_name=item_name, quantity=quantity)}
    
  3. 如果您愿意,可以尝试对其他路线进行类似的替换。否则,您可以将文件的全部内容替换为以下行:

    import redis
    from fastapi import FastAPI, HTTPException
    
    from models import ItemPayload
    
    app = 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 exists
        item_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 item
            item_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 too
            redis_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_str
            else:
                continue  # skip this item if it has no name
    
            item_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 = 0
    
            items.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 item
        if item_quantity is None:
            existing_quantity: int = 0
        else:
            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."}
    
    
  4. 重新运行调试器以通过与/docs路由交互来测试该应用程序。完成后您可以停止调试器。

恭喜!现在,您拥有一个可运行的 FastAPI 应用程序,其中包含用于在购物清单中添加、列出和删除商品的路由,并且数据保存在 Redis 数据库中。

可选:设置数据库删除

现在数据已由 Redis 保存,您可能需要创建一个脚本来删除所有测试数据。为此,请创建一个名为的新文件,flushdb.py其中包含以下内容:

import redis

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

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

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

可选:创建 ChatGPT 插件

借助 GitHub Codespaces,您可以在使用ChatGPT 插件时托管应用程序以进行测试。ChatGPT 插件是使ChatGPT能够与现有 API 交互的工具,以增强 ChatGPT 的能力,使其能够执行各种操作。ChatGPT 插件目前尚未公开提供,但您可以加入他们的等待名单以获取访问权限。完成后,您可以按照下面的实时流录制为 ChatGPT 创建您自己的购物清单插件:

注意:所有个人 GitHub.com 帐户都有每月免费使用免费或专业计划中包含的 GitHub Codespaces 的配额。有关更多信息,请转到关于 GitHub Codespaces 的计费

下一步

感谢您遵循本教程!我们希望您了解有关 FastAPI 的新知识以及如何将其与 VS Code 结合使用。

本教程完成的代码项目可以在 GitHub 上找到:python-sample-vscode-fastapi-tutorial

请参阅官方文档了解有关 FastAPI 的更多信息。

要在生产网站上试用该应用程序,请查看教程使用 Docker 容器将 Python 应用程序部署到 Azure 应用服务

您还可以查看其他 VS Code Python 文章: