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中进行开发,您可以在代码空间中远程编码、调试和运行您的应用程序。代码空间提供了一个完全配置的云端开发环境,消除了本地设置的需求。此环境包括您项目的依赖项、工具和扩展,确保了一致且可复现的开发体验。它通过提供实时编辑、集成版本控制以及轻松访问调试和测试工具来简化协作,同时保持项目的安全性和可靠性。

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

要为本教程设置一个代码空间,请导航到此项目的GitHub仓库。此代码空间包含所有必要的配置和依赖项,以便快速开始FastAPI开发。

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

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

然后,选择代码 > 代码空间 > 分支上创建代码空间,为您的项目创建并打开一个代码空间。

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

在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

    下拉菜单显示“Venv”或“Conda”作为可以使用Python: Create Environment命令创建的环境选项

  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

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

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

    浏览器中显示的Hello World消息

    恭喜!您的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类型检查模式”并将其设置为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将请求路由到适当的函数(也称为路由处理程序或视图函数),该函数处理请求并生成响应。

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

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

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

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

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

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

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

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

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

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

    "试试看"按钮显示在Swagger UI中的/items路由旁边

  7. 通过向item_name字段传递一个字符串和向quantity字段传递一个数字来添加一个杂货清单项。例如,你可以提供苹果作为item_name,并提供2作为quantity

  8. 选择执行

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

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

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

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

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

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

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

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

    这将打开调试控制台并运行选定的表达式。正如我们示例中所预期的,表达式评估为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 的 Dev Containers 扩展提供了一种简化的方法,将您的项目、其依赖项和所有必要的工具整合到一个整洁的容器中,创建一个功能齐全的开发环境。该扩展允许您在 VS Code 中打开容器内(或挂载到容器中)的项目,您将拥有其完整的功能集。

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

需求

创建开发容器配置

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

  2. 选择 Python 3:

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

  3. 选择默认版本。

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

    我们可以选择性地安装Features以包含在容器中。对于本教程,我们将安装Redis Server,这是一个社区贡献的Feature,它安装并为Redis添加了适当的开发容器设置。

    在Dev Containers配置文件列表中选择的Redis Server选项

    这会在您的工作区中创建一个.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. 从右下角显示的通知中选择在容器中重新打开,或从命令面板运行Dev Containers: 在容器中重新打开命令。

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

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

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

一旦容器设置完成,你会在VS Code的左下角看到一个指示器:

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

注意: 通过打开扩展视图(⇧⌘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_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 尚未被导入。

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

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

    提示: 你可以通过查找设置编辑器中的自动导入补全设置(⌘, (Windows, Linux Ctrl+,))并启用它来设置Pylance自动添加导入。

    我们现在有一个连接到本地主机上运行的Redis服务器的Redis客户端对象(host="0.0.0.0"),并在端口6379上监听(port=6379)。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_idint 类型。让我们通过重命名 item_id 符号来解决这个警告。

  4. item_id重命名为item_id_str

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

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

  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方法将项目添加到Redis哈希中,并为字段(item_iditem_namequantity)和值(项目的新创建的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,
                })
    

    现在我们只需要通过设置我们最初引用的哈希值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: 在终端中运行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文章: