在 VS Code 中试用

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 账户都包含免费或 Pro 计划中的 GitHub Codespaces 每月免费使用配额。有关更多信息,请参阅关于 GitHub Codespaces 的计费

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

对于本教程,请选择 dictionarybased 分支

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

然后,选择 Code > 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: Create Environment 命令来创建虚拟环境。

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

  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: Select Interpreter 命令)并手动选择虚拟环境。

开始编写代码

让我们创建应用程序!

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

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

  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

    提示:如果您的默认端口已被占用,请停止调试器并打开命令面板(⇧⌘P(Windows、Linux Ctrl+Shift+P),搜索 Debug: Add Configuration,选择 Python Debugger,然后选择 FastAPI。这将在 .vscode/launch.json 中创建一个您可以编辑的自定义配置文件。将以下内容添加到 "args":[] 以设置自定义端口:"--port=5000"。保存文件,并使用(F5)重新启动调试器。

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

    Hello World message displayed in the browser

    恭喜!您的 FastAPI 应用已成功运行!

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

为购物清单项创建模型

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

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

  1. 使用文件 > 新建文件... 创建一个新的 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 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 将请求路由到适当的函数(也称为路由处理程序或视图函数),然后该函数处理请求并生成响应。

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

  3. 将以下路由添加到 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 应用程序中定义的元数据和类型提示动态生成。

  4. 通过单击行号左侧空白处(或按 F9),在 if quantity <= 0 语句旁边添加一个断点。调试器将在此行执行之前停止,以便您可以逐行检查代码。

    Breakpoint set next to the first line in the add_item function

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

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

    Swagger UI displaying two endpoints: /items and /

  6. 选择 /items 路由旁边的向下箭头将其展开,然后选择右侧出现的试一试按钮。

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

  7. 通过向 item_name 字段传递一个字符串,向 quantity 传递一个数字来添加一个购物清单项。例如,您可以将 'apple' 作为 item_name,将 2 作为 quantity

  8. 选择执行

    Execute button displayed below the /items route

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

    Debugger stopped at the breakpoint set in the add_item function

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

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

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

  10. 选择 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来继续执行代码。

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

  11. 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."}
    
    
  12. 保存文件(⌘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 的开发容器扩展提供了一种简化的方法,可以将您的项目、其依赖项以及所有必需的工具整合到一个整洁的容器中,创建一个功能齐全的开发环境。该扩展允许您在 VS Code 中打开容器内(或挂载到容器内)的项目,在那里您将拥有其全部功能集。

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

要求

创建开发容器配置

  1. 打开命令面板并运行开发容器: 添加开发容器配置文件...

  2. 选择Python 3

    Python 3 option selected in the Dev Containers configuration files list

  3. 选择默认版本。

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

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

    Redis Server option selected in the Dev Containers configuration files list

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

  5. 打开 devcontainer.json 文件。

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

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

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

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

    您可以在开发容器规范中了解更多关于 postCreateCommand 和其他生命周期脚本的信息。

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

  8. 将以下设置添加到 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
                ]
            }
        }
    
  9. 保存文件。

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

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

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

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

容器设置完成后,您会在 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)
    

    由于尚未导入 Redis,Pylance 将显示错误消息。

  2. 将光标放在编辑器中的“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

  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 方法返回 strNone(如果项目不存在)。但是,我们尚未替换的代码下方的行期望 item_id 的类型是 int。让我们通过重命名 item_id 符号来解决此警告。

  4. item_id 重命名为 item_id_str

  5. 如果已启用内联提示 (inlay hints),Pylance 应该在 item_id_str 旁边显示一个变量类型提示。可以选择双击接受它。

    Variable type hint displayed next to the item_id_str variable

  6. 如果项目不存在,则 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 字段。

  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_strNone 时。在这种情况下,我们将生成一个新的 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_iditem_namequantity)提供映射以及值(项目新创建的 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,
                })
    

    现在,我们只需通过设置我们在开头引用的 item_name_to_id 哈希,将新创建的 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, 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."}
    
    
  13. 重新运行调试器,通过与 /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 Plugins 时,可以托管您的应用程序用于测试目的。ChatGPT Plugins 是一种工具,使 ChatGPT 能够与现有 API 交互,从而增强 ChatGPT 的能力,使其能够执行广泛的操作。ChatGPT Plugins 当前尚未公开可用,但您可以加入其等待列表以获取访问权限。一旦获得访问权限,您可以跟着下面的直播录像操作,为 ChatGPT 创建自己的购物清单插件:

注意:所有个人 GitHub.com 账户都包含在 Free 或 Pro 计划中,具有每月免费使用 GitHub Codespaces 的额度。有关更多信息,请参阅About billing for GitHub Codespaces

下一步

感谢您跟随本教程!希望您学到了一些关于 FastAPI 以及如何在 VS Code 中使用它的新知识。

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

官方文档中了解有关 FastAPI 的更多信息。

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

您还可以查阅这些其他 VS Code Python 文章: