Visual Studio Code 中的 FastAPI 教程

FastAPI 是一个现代高性能的网络框架,用于用 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 Codespace 中设置这个项目,在那里你可以在 Codespace 中远程编程、调试和运行你的应用。代码空间提供了一个完全配置的开发环境,托管在云端,无需本地搭建。该环境包含项目的依赖、工具和扩展,确保开发体验一致且可重复。它通过提供实时编辑、集成版本控制以及便捷的调试和测试工具访问,简化了协作流程,同时保持项目的安全性和可靠性。

注意:所有 GitHub.com 账户在免费或专业套餐中均有每月免费使用GitHub Codespaces的配额。欲了解更多信息,请访问 GitHub Codespaces 的计费

要为本教程设置代码空间,请访问该项目的 GitHub 仓库。该代码空间包含所有快速启动 FastAPI 开发所需的配置和依赖。

在本教程中,选择基于词典的分支:

DictionaryBased 分支在 Python Sample-VScode-FastAPI 教程 GitHub 仓库中选定

然后,选择代码>代码空间,>在<基于字典的分支创建代码空间>分支,创建并打开你的项目代码空间。

完成后,你可以继续作下面的“替换数据库”部分。

本地 VS Code 中

要成功完成这个VS Code教程,首先你需要搭建你的Python开发环境。具体来说,这个教程要求:

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

  1. 在你的文件系统中,为这个教程创建一个项目文件夹,比如杂货插件.

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

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

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

我们将安装 FastAPI 来创建应用,uvicorn 作为服务器运行,以及 RedisType-redis(类型-redis用于处理数据存储和与 Redis 数据库交互。

  1. 在 VS Code 中创建新文件(文本文件>文件,Windows,Linux 按 Ctrl+N⌘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)在杂货插件文件夹。

  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。

    提示:如果你的默认端口已经在使用,停止调试器,打开命令面板(⇧⌘P(Windows,Linux Ctrl+Shift+P),搜索“调试:添加配置”,选择 Python 调试器,然后选择 FastAPI。这会创建一个自定义配置文件.vscode/launch.json你可以编辑。添加以下内容“ARGS”:[]设置自定义端口:“--port=5000”.保存文件,然后用 (F5) 重启调试器。

  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 请求中自动验证和解析输入数据(称为“payloads”)。

让我们为购物清单上的物品创建一个模型。我们将使用ItemPayload模型定义要添加到购物清单中的物品数据结构。该模型将包含三个字段:item_id,item_name, 和数量.

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

  2. 将以下几行添加到文件中,然后保存在杂货插件文件夹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 是 Python VS Code 的默认语言服务器,支持类型提示功能,有助于处理 Pydantic 模型和 FastAPI。这是因为Pylance建立在Pyright之上,Pyright是一个静态的Python类型检查工具,可以检测代码中的类型错误,防止错误并提升代码质量。

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

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

  2. 搜索“python类型检查模式”,并将其设置为基础用于基本类型检查。Pylance现在将显示诊断和警告,以捕捉简单的类型相关错误。或者,你也可以设置为严格以执行更高级的类型检测规则

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

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

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

创建路由

现在我们需要一个地方来存放购物清单上的物品。为了简化,我们先从一个空字典开始。

  1. 首先,导入样本所需的所有包。打开main.py文件并用以下格式替换第一行导入:

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

    grocery_list: dict[int, ItemPayload] = {}
    

    这会创建一个新的空字典,用于接收类型为 的键智力(作为项目ID)以及ItemPayload类型。

    接下来我们将在FastAPI应用中定义路由。在网页应用的语境中,路由就像将特定URL映射到处理它们的代码的路径。这些路由作为应用中不同功能的入口。当客户端(如网页浏览器或其他程序)向我们的应用程序发送带有特定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_ids以及item_id.你可以选择双击每个建议,将它们插入代码中:

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

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

  4. 如果数量<= 0通过点击行号左边缘(或F9)即可。调试器会在执行该行前停止,所以你可以逐行检查代码。

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

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

    应用中应该有一个Swagger接口,包含两个端点:/条目根()。/

    Swagger 界面显示两个端点:/items 和 /

  6. 选择旁边的下箭头/条目通过路线展开,然后点击右侧出现的“试用”按钮。

    在Swagger界面的/items路线旁边,显示了“试试”按钮

  7. 通过传递字符串到item_name域和一个数 到数量.例如,你可以提供 apple 作为item_name2 作为数量.

  8. 选择执行

    执行按钮显示在/items路线下方

  9. 再次打开 VS Code,发现调试器已经停在你之前设定的断点。

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

    在左侧,此时定义的所有局部和全局变量显示在变量窗口的运行和调试视图下。在我们的例子中,item_name设置为“apple”,且数量在 locals 变量视图下设置为 2,并且是空的grocery_list全局变量视图下的词典。

    变量窗口显示在运行和调试视图中,项目和grocery_list变量被高亮显示

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

  10. 选择数量 <= 0右键点击编辑器,选择调试控制台中的评估

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

    这会打开调试控制台并运行所选表达式。如我们示例中预期的那样,该表达式的值为错误.

    调试控制台可以是一个强大的工具,可以快速测试表达式,更好地理解断点时代码的状态。你也可以用它运行任意代码,比如调用函数或打印变量。你可以在 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)。应用程序应该会自动重新加载。

你现在可以打开/文档再次页面并测试新路由,使用调试器和调试控制台更好地理解代码执行情况。完成后,你可以停止调试器(⇧F5(Windows,Linux Shift+F5)。你也可以点击第4步添加的断点来移除它。

恭喜你!你现在拥有了一个可用的FastAPI应用程序,可以路由添加、列出和删除购物清单中的物品。

设置数据存储

此时,你已经有一个带有基础功能的可用应用版本。本节指导你如何设置数据存储以实现持久化,但如果你对已经学到的内容满意,也可以选择跳过。

目前我们把数据存储在字典里,这并不理想,因为应用重启时所有数据都会丢失。

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

如果你已经在用我们现有模板在 GitHub Codespaces 上工作,可以直接跳到“替换数据库”部分。

如果你用的是Windows,可以通过搭建Docker容器GitHub代码空间来作Redis。本教程将使用Docker容器,但你可以参考上面部分了解如何搭建GitHub代码空间。

否则,如果你用的是Linux或macOS机器,可以按照他们官网的说明安装Redis,然后跳到“替换数据库”部分。

在Windows上搭建Docker容器

VS Code 开发容器扩展提供了一种简化的方式,将你的项目、依赖以及所有必要工具整合到一个整洁的容器中,打造一个功能齐全的开发环境。这个扩展允许你在 VS Code 中打开(或挂载到)容器中,拥有完整的功能集。

以下步骤请确保你的机器安装了以下要求:

要求

创建开发容器配置

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

  2. 选择 Python 3

    在开发容器配置文件列表中选择的Python 3选项

  3. 选择默认版本。

  4. 选择 Redis 服务器作为安装的附加功能,点击确定,然后选择“保留默认值”。

    我们可以选择安装包含在容器中的特征。在这个教程中,我们将安装Redis Server,这是一个由社区贡献的功能,用于安装和添加Redis所需的开发容器设置。

    在开发容器配置文件列表中选择的Redis Server选项

    这会生成一个.devcontainer在你的工作区里,带有一个devcontainer.json档案。我们对这个文件做些编辑,使容器设置包括安装所需的 VS Code 扩展以及项目依赖等步骤。

  5. 打开devcontainer.json档案。

  6. “特色”:{ ... }这样我们就能在文件中添加更多设置。

    接下来,我们将在postCreateCommand财产在devcontainer.json文件,这样我们的应用一旦容器设置好就可以运行了。

  7. 找到下面的内容,并从该行移除注释(),这样在容器创建后就可以安装依赖://

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

    你可以了解postCreateCommand以及更多生命周期脚本,包含在开发容器规范中。

    现在我们将使用自定义属性来添加我们希望安装在容器中的 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. 从右下角显示的通知中选择“在容器中重新打开”,或者在命令面板中执行“开发容器:在容器中重新打开”命令。

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

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

完成后,你将拥有一个完整配置的基于Linux的工作空间,安装了Python 3和Redis Server。

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

开发容器指示器显示在VS Code的左下角

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

所选的 Python 解释器信息可在右下角的状态栏中查看,与devcontainer.json文件:

Python 解释器选择

注意:如果你在状态栏找不到 Python 解释器的信息,可以点击 Python 解释器指示器(或从命令调色板中运行 Python: 选择解释器命令),手动选择容器中的 Python 解释器。

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

替换数据库

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

与传统数据库中可以不了解其ID即可检索项目不同,你需要知道Redis的哈希键才能从中获取值。在本教程中,我们将创建一个名为item_name_to_id通过名称检索项目,并将其映射到其ID。此外,我们还会创建其他哈希值,按ID检索项目,映射到它们的名称和数量。每个项目哈希都有名称item_id:{item_id}并且有两个域:item_name以及数量.

首先,我们先用连接到 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”上,点击显示的灯泡(或⌘。(Windows,Linux Ctrl+))。然后选择添加“导入redis”

    灯泡显示在 Redis 变量旁边,并有添加导入语句的选项

    提示:你可以设置Pylance自动添加导入,方法是在设置编辑器(⌘,(Windows,Linux Ctrl+,)中找到自动导入完成设置并启用。

    我们现在有一个 Redis 客户端对象,连接到运行在本地主机上的 Redis 服务器 (host=“0.0.0.0”)以及在6379端口监听(port=6379). 该DB参数指定了要使用的 Redis 数据库。Redis 支持多个数据库,在这段代码中我们将使用 Database 0,这是默认数据库。我们还在传递decode_responses=真以字符串(而非字节)来解码响应。

    我们先在第一条路上做更多替换add_item. 我们不必从词典中查看所有密钥来查找已提供的项目名称,而是直接从 Redis 哈希中获取这些信息。

    我们假设item_name_to_id哈希已经存在,将物品名称映射到它们的ID(别担心,我们很快会添加这个代码!)。然后我们可以通过调用来自 Redis 的方法,如果请求的名称已存在于哈希中,则返回该项目 ID,或者没有如果不行的话。

  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对这一变化提出了问题。这是因为方法返回的力量,或者没有(如果该项不存在的话)。然而,代码下方尚未替换的行预计会item_id为类型智力. 让我们通过重新命名item_id象征。

  4. 更名item_iditem_id_str.

  5. 如果你启用了内嵌提示,Pylance应该会在旁边显示一个可变类型的提示item_id_str. 你可以选择双击接受:

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

  6. 如果该物品不存在,那么item_id_str没有. 所以现在我们可以删除包含以下内容的那条线:

    if item_name in items_ids.keys():
    

    并用:

    if item_id_str is not None:
    

    现在我们有了 item ID 作为字符串,需要将其转换为智力并更新该物品的数量。目前,我们的 Redis 哈希仅将物品名称映射到 ID 上。为了将物品 ID 映射到它们的名称和数量,我们将为每个物品创建一个单独的 Redis 哈希,使用“item_id:{item_id}”作为我们的哈希名,以便通过ID检索更方便。我们还会添加item_name以及数量每个哈希值的字段。

  7. 删除如果阻挡:

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

    并添加以下条件,以便转换为item_id到一个智力然后通过调用欣克比来自Redis的方法。该方法递增“数量”字段 以请求中给定的金额 (数量):

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

    现在只需替换当该项不存在时的代码,当item_id_str没有. 在这种情况下,我们生成一个新的item_id为该物品创建一个新的 Redis 哈希值,然后添加提供的物品名称和数量。

    生成新的item_id,我们用Incr来自Redis的方法,传递一个新的哈希值,称为“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")
    

    当这时Incr调用首次运行时使用item_idskey,Redis创建key并将其映射到值1. 然后,每次运行时,都会将存储值增加1。

    现在我们将将该项添加到 Redis 哈希中,使用hset通过为这些字段(item_id,item_name, 和数量),以及这些值(物品新创建的ID,以及其提供的名称和数量)。

  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. 将这条线添加到路线终点,在否则阻挡:

    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. 重新运行调试器以测试该应用程序,与/文档路线。完成后你可以停止调试器。

恭喜!你现在有一个可用的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: Run Python File in Terminal 命令。

请注意,这样做时应谨慎,因为会删除当前数据库中的所有密钥,如果在生产环境中执行,可能会导致数据丢失。

可选:创建GPT动作

通过GitHub Codespaces,你可以托管应用进行测试,使用GPT Actions。GPT Actions是工具,使ChatGPT能够与现有API交互,增强ChatGPT的能力,使其能够执行多种作。你可以跟随下面的直播录制,为ChatGPT创建你自己的购物清单插件:

注意:所有个人 GitHub.com 账户的免费或专业计划均包含每月免费使用GitHub Codespaces的配额。欲了解更多信息,请访问关于GitHub Codespaces计费

下一步

感谢你跟随本教程!希望你学到了关于FastAPI以及如何与VS Code结合使用的新知识。

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

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

想在生产环境中试用应用,可以看看教程《使用Docker Containers部署Python应用到Azure App Service》。

你也可以查看这些其他VS Code Python文章: