现已推出!阅读 10 月份的新功能和修复。

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 Codespaces

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

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

要为本教程设置 codespace,请导航到此项目的 GitHub 存储库。此 codespace 包含所有必要的配置和依赖项,以便快速开始 FastAPI 开发。

在本教程中,请选择dictionarybased 分支

dictionarybased branch selected in the python-sample-vscode-fastapi-tutorial GitHub repo

然后,选择代码 > Codespaces > 在 <dictionarybased> 分支上创建 Codespace 以创建并打开项目的 codespace。

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

本地 VS Code

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

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

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

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

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

现在,让我们创建一个 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: 创建环境命令来创建虚拟环境。

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

  5. 当提示您选择环境类型时,请选择Venv

    Dropdown with "Venv" or "Conda" as options for environments that can be created with the Python: Create Environment command

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

    List of available global environments that can be used to create a virtual environment

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

    Check box selected to install dependencies from requirements.txt file

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

Environment in the Status bar

注意:如果您在状态栏上找不到新创建的环境信息,您可以单击 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配置选项。

    Dropdown with debugger configuration options, with FastAPI being highlighted

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

    Uvicorn server running message displayed in the terminal, with an URL to access the app

  6. Ctrl+Click终端中的http://127.0.0.1:8000/ URL,在您的默认浏览器中打开该地址。

    Hello World message displayed in the browser

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

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

为杂货清单项目创建模型

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

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

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

  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 type checking mode”并将其设置为basic以进行基本类型检查。现在,Pylance 将显示诊断和警告以捕获简单的类型相关错误。或者,您可以将其设置为strict以执行更多高级的类型检查规则

    Python Analysis Type Checking Mode options (off, basic and strict) in Settings editor

  3. 接下来,搜索“Python inlay type hints”,并为变量类型函数返回值类型启用内联提示。

    Two Python Analysis Type Hints settings being enabled in the Settings editor: for Function Return Types and for Variable Types

创建路由

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

  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_idsitem_id的类型。您可以选择双击每个建议以将其插入代码中。

    Inlay function return and variable type hints being displayed by Pylance throughout the sample code

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

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

    Breakpoint set next to the first line in the add_item function

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

    应该有一个 Swagger 界面,其中包含应用程序中可用的两个端点:/items和根目录(/)。

    Swagger UI displaying two endpoints: /items and /

  3. 选择/items路由旁边的向下箭头以展开它,然后选择右侧出现的试用它按钮。

    "Try it out" button displayed next to the /items route in the Swagger UI

  4. 通过将字符串传递给item_name字段和将数字传递给quantity来添加一个购物清单项目。例如,您可以提供 apple 作为item_name,并提供 2 作为quantity

  5. 选择执行

    Execute button displayed below the /items route

  6. 再次打开 VS Code 并注意调试器已停止在您之前设置的断点处。

    Debugger stopped at the breakpoint set in the add_item function

    在左侧,在运行和调试视图下的变量窗口中,显示了此时定义的所有局部和全局变量。在我们的示例中,item_name设置为'apple',quantity设置为 2,位于 locals 变量视图中,以及一个空的grocery_list字典,位于 globals 变量视图中。

    Variables window displayed in the Run and Debug view, with the item and grocery_list variables highlighted

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

  1. 选择quantity <= 0语句,右键单击编辑器并选择在调试控制台中评估

    Evaluate in Debug Console option displayed in the context menu when right-clicking on a line of code

    这将打开调试控制台并运行所选表达式。正如我们在示例中预期的那样,表达式评估结果为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来使用 Redis。在本教程中,我们将使用 Docker 容器,但您可以参考上面的部分以获取有关如何设置 GitHub Codespace 的说明。

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

在 Windows 上设置 Docker 容器

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

对于以下步骤,请确保您已在您的机器上安装了以下要求。

要求

创建 Dev 容器配置

  1. 打开命令面板并运行Dev Containers: Add Dev Container Configuration Files…

  2. 选择Python 3

    Python 3 option selected in the Dev Containers configuration files list

  3. 选择默认版本。

我们可以选择安装功能到容器中。在本教程中,我们将安装Redis 服务器,这是一个社区贡献的功能,用于安装和添加 Redis 的正确开发容器设置。

  1. 选择**Redis 服务器**作为要安装的附加功能,点击**确定**,然后选择**保持默认值**。

    Redis Server option selected in the Dev Containers configuration files list

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

  1. 打开devcontainer.json文件。

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

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

  1. 找到下面的内容,删除该行的注释(//),这样在创建容器后就可以安装依赖项了

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

    您可以在开发容器规范中了解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: Reopen in Container**命令。

    **注意**:根据网络速度和机器性能,构建容器可能需要几分钟时间。

    您可以在Dev Containers 文档中了解更多关于开发容器配置的信息。

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

容器设置好后,您会在 VS Code 的左下角看到一个指示器

Dev Containers indicator displayed on the bottom left corner of VS Code

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

选定的 Python 解释器信息在右下角的状态栏上可用,与devcontainer.json文件中指定的版本匹配

Python interpreter selection

**注意**:如果您在状态栏上找不到 Python 解释器信息,您可以点击 Python 解释器指示器(或从命令面板运行**Python: Select Interpreter**命令),然后在容器中手动选择 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+.))。然后选择**添加“import redis”**。

    Light bulb displayed next to the Redis variable, with the option to add the import statement

    **提示**:您可以通过在设置编辑器 (⌘, (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 的hget方法来获取我们正在请求中接收的项目名称的 ID,如果请求的名称在哈希中已经存在,则该方法将返回项目 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方法返回strNone(如果项目不存在)。但是,我们尚未替换的代码下面的行期望item_id的类型为int。让我们通过重命名item_id符号来解决此警告。

  1. item_id重命名为item_id_str

  2. 如果启用了内联提示,Pylance 应该会在item_id_str旁边显示一个变量类型提示。您也可以选择双击以接受它

    Variable type hint displayed next to the item_id_str variable

  3. 如果项目不存在,则item_id_strNone。所以现在我们可以删除具有以下内容的行

    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_namequantity字段。

  1. 删除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_strNone时。在这种情况下,我们生成一个新的item_id,为项目创建一个新的 Redis 哈希,然后添加提供的项目名称和数量。

为了生成一个新的item_id,让我们使用 Redis 的incr方法,传递一个名为"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")
    

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

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

  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()

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

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

可选:创建 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 文章