使用 OpenAPI 规范 （Swagger） 设置 Microsoft Fabric 工作负荷后端

Microsoft Fabric 工作负荷后端是实现 Fabric API 协定的服务，使自定义工作负载能够与 Microsoft Fabric 平台无缝集成。 此后端处理工作负荷项的生命周期作，包括创建、检索、更新和删除。

本文指导你直接从 OpenAPI （Swagger） 定义快速生成 Fabric 工作负荷后端。 借助此 API 优先方法，即使在将后端逻辑集成到完整的Microsoft Fabric 开发环境之前，也能快速对后端逻辑进行原型制作和验证。 此处演示的原则非常适用，无论选择哪种特定工具或语言。

学完本文后，你将能够：

• 基于示例中包含的 Swagger 文件生成 Fabric 工作负荷后端。
• 了解 Fabric 工作负载后端的基本结构和组件。
• 使用 Python 和 FastAPI 在本地运行和测试生成的后端。

在本文中，你将实现项目生命周期内的以下核心操作。 这些操作与 Fabric API Swagger 文件中定义的终结点相对应。

• 创建项：初始化新的工作负荷项。
• 获取项有效负载：检索项配置。
• 更新项：修改现有项。
• 删除项：从工作区中删除项。

本文专门演示了将 Python 和 FastAPI 与 OpenAPI 生成器工具结合使用的过程。 但是，OpenAPI 生成器本身支持 许多编程语言和框架。 你可以自由选择任何符合团队专业知识和项目需求的 OpenAPI 兼容的代码生成工具或方法，以创建后端框架。

先决条件

在开始执行文章中的步骤之前，请确保您拥有以下项目。

必需的知识

• 了解 Microsoft Fabric 项目生命周期。 阅读 项生命周期管理。

  本文对此理解至关重要。 生成的后端实现了 Fabric 项的生命周期作业（创建、读取、更新、删除），如 项生命周期文档中所定义。

• Python 和 RESTful API 的基础知识。

• 熟悉 Microsoft Fabric 工作负荷概念。

所需软件

• Python 3.8+。 下载 Python。
• Node.js，如果要通过 npm 安装 OpenAPI 生成器 CLI，则需要该 CLI。 下载 Node.js。
• Git，用于克隆示例存储库。 下载 Git。
• 代码编辑器，如 Visual Studio Code、PyCharm 或首选集成开发环境（IDE）。

适用于 OpenAPI 生成器的 Java

OpenAPI 生成器 CLI 需要 Java 作为运行时环境。 无需编写 Java 代码。 你只需要它来运行生成器工具。

所需的最低 Java 版本是 Java 8。 建议使用受支持的长期支持（LTS）版本，例如 Java 17 或 Java 21。

安装 Java：

1. 安装 OpenJDK 的Microsoft版本（建议）。 按照您的操作系统，在安装 Microsoft 内部版本的 OpenJDK中的说明进行操作。

2. 验证安装。 打开终端或命令提示符并运行：

   java -version
   

   此时应会看到如下例所示的输出：

   openjdk version "17.0.12" 2024-07-16 LTS
   OpenJDK Runtime Environment Microsoft-10377968 (build 17.0.12+7-LTS)
   OpenJDK 64-Bit Server VM Microsoft-10377968 (build 17.0.12+7-LTS, mixed mode, sharing)
   

如果已从另一个供应商（例如 Oracle、Eclipse Temurin 或 Amazon Corretto）安装了 Java（例如版本 8 或更高版本），则可以使用现有安装。

步骤 1：设置开发环境

首先，使用所需的工具和包设置开发环境：

1. 克隆 Microsoft Fabric 开发样本库：

   git clone https://github.com/microsoft/Microsoft-Fabric-workload-development-sample
   cd Microsoft-Fabric-workload-development-sample
   

2. 创建 PythonBackend 目录：

   mkdir PythonBackend
   cd PythonBackend
   

3. 创建 Python 虚拟环境：

   # Create a Python virtual environment for the project
   python -m venv .venv
   
   # Activate the virtual environment
   # Windows
   .venv\Scripts\activate
   
   # macOS/Linux
   source .venv/bin/activate
   

4. 安装 OpenAPI 生成器 CLI：

   npm install @openapitools/openapi-generator-cli -g
   

   有关替代安装方法，请参阅 OpenAPI 生成器安装文档。

步骤 2：验证 Python 虚拟环境是否处于活动状态

创建虚拟环境后，请务必确保使用正确的 Python 解释器。 此方法使项目依赖项保持隔离并正确管理。

验证虚拟环境的激活

确认虚拟环境已激活。 应在终端提示符的开头看到 (.venv) 。

如果未激活虚拟环境，请运行：

# Windows
.venv\Scripts\activate

# macOS/Linux
source .venv/bin/activate

验证虚拟环境的 Python 解释器是否处于活动状态

确认终端正在使用虚拟环境中的 Python 解释器，而不是系统的全局 Python 安装。

运行下面的命令：

# Display the path to the active Python interpreter
python -c "import sys; print(sys.executable)"

预期输出应指向虚拟环境：

- Windows: C:\path\to\project\PythonBackend\.venv\Scripts\python.exe
- macOS/Linux: /path/to/project/PythonBackend/.venv/bin/python

配置 IDE （可选）

大多数新式 IDE 会自动检测 Python 虚拟环境。 但是，可能需要在 IDE 设置中手动选择解释器。

排查虚拟环境问题

始终确保在安装依赖项或运行应用程序之前激活虚拟环境。 (.venv)终端中的前缀确认激活状态。 如果遇到导入错误或缺少包，请通过运行前面提到的验证命令来验证是否正在使用正确的 Python 解释器。

步骤 3：从 OpenAPI 规范生成 FastAPI 项目

使用 OpenAPI 生成器 CLI 从 Fabric API 的 Swagger 规范创建 Python FastAPI 项目。

运行生成命令

从 PythonBackend 目录中运行以下命令：

openapi-generator-cli generate -i ../Backend/src/Contracts/FabricAPI/Workload/swagger.json -g python-fastapi -o . --additional-properties=packageName=fabric_api

此命令指示 OpenAPI 生成器 CLI 执行以下作：

¹ 对于 [InputSpecPath]，路径为 ../Backend/src/Contracts/FabricAPI/Workload/swagger.json。

² 对于生成器 （-g） 参数，本文使用值 python-fastapi 作为示例。 OpenAPI 生成器支持多种语言和框架的服务器端代码生成器。 您可以将 python-fastapi 替换为您想要的生成器。 有关全面的列表，请参阅 OpenAPI 服务器生成器文档。

安装所需的依赖项

若要安装依赖项，请使用以下命令：

pip install -r requirements.txt

在 Windows 上，您可能会遇到 uvloop 包的错误。 如果发生这种情况：

1. 打开 requirements.txt 文件。

2. 查找 uvloop 条目，它可能看起来类似于 uvloop==0.17.0。 将平台条件添加到末尾：

   uvloop==<existing version>; sys_platform != 'win32'
   

   例如，如果文件具有 uvloop==0.17.0，请将其更改为 uvloop==0.17.0; sys_platform != 'win32'。

3. 再次运行 pip install -r requirements.txt。

此更改可确保 uvloop 仅在非 Windows 平台上安装。

步骤 4：了解生成的代码结构

OpenAPI 生成器使用以下密钥目录创建结构化 FastAPI 项目：

PythonBackend/
├── src/
│   └── fabric_api/
│       ├── apis/              # Generated API route definitions
│       │   ├── item_lifecycle_api.py
│       │   ├── jobs_api.py
│       │   └── endpoint_resolution_api.py
│       ├── impl/              # Where you'll implement controllers
│       │   └── __init__.py
│       ├── models/            # Data models for requests/responses
│       │   ├── create_item_request.py
│       │   └── ...
│       └── main.py            # FastAPI application entry point
├── tests/                     # Generated test files
└── requirements.txt           # Dependencies

• 该 apis 目录包含每个 API 终结点的路由器定义。
• 该 models 目录包含用于请求和响应对象的 Pydantic 模型。
• 目录 impl 用于实现控制器逻辑。
• 该文件 main.py 设置 FastAPI 应用程序。

步骤 5：实现 ItemLifecycle 控制器

创建处理 Fabric API 请求的控制器实现。 控制器继承自生成的基类。

在impl目录中创建item_lifecycle_controller.py：

# file path: src/fabric_api/impl/item_lifecycle_controller.py

from fabric_api.apis.item_lifecycle_api_base import BaseItemLifecycleApi
from fabric_api.models.get_item_payload_response import GetItemPayloadResponse
from pydantic import Field, StrictStr
from typing_extensions import Annotated
from fastapi import HTTPException

class ItemLifecycleController(BaseItemLifecycleApi):
    """
    Implementation of Item Lifecycle API methods.
    """
    
    async def item_lifecycle_create_item(
        self,
        workspaceId,
        itemType,
        itemId,
        activity_id,
        request_id,
        authorization,
        x_ms_client_tenant_id,
        create_item_request,
    ) -> None:
        """
        Implementation for creating a new item.
        """
        print(f"\n=== CREATE ITEM CALLED ===")
        print(f"Workspace ID: {workspaceId}")
        print(f"Item Type: {itemType}")
        print(f"Item ID: {itemId}")
        print(f"Display Name: {create_item_request.display_name}")
        print(f"Description: {create_item_request.description}")
        if create_item_request.creation_payload:
            print(f"Creation Payload: {create_item_request.creation_payload}")
        print("===========================\n")
        return
    
    async def item_lifecycle_delete_item(
        self,
        workspaceId,
        itemType,
        itemId,
        activity_id,
        request_id,
        authorization,
        x_ms_client_tenant_id,
    ) -> None:
        """
        Implementation for deleting an existing item.
        """
        print(f"Delete item called for itemId: {itemId}")
        return
    
    async def item_lifecycle_get_item_payload(
        self,
        workspaceId,
        itemType,
        itemId,
        activity_id,
        request_id,
        authorization,
        x_ms_client_tenant_id,
    ) -> GetItemPayloadResponse:
        """
        Implementation for retrieving the payload for an item.
        """
        print(f"Get item payload called for itemId: {itemId}")
        # Return a simple payload
        return GetItemPayloadResponse(item_payload={"sample": "data"})
    
    async def item_lifecycle_update_item(
        self,
        workspaceId,
        itemType,
        itemId,
        activity_id,
        request_id,
        authorization,
        x_ms_client_tenant_id,
        update_item_request,
    ) -> None:
        """
        Implementation for updating an existing item.
        """
        print(f"Update item called for itemId: {itemId}")
        return

步骤 6：配置并运行 FastAPI 应用程序

在运行 FastAPI 应用程序之前，请确保端口配置与 Microsoft Fabric 开发环境保持一致。 此步骤对于正确与 Fabric 开发网关集成至关重要。

了解端口配置

开发 Microsoft Fabric 工作负荷时，开发网关会将 API 请求路由到后端。 此配置需要：

• 要在特定端口上运行的后端（默认值：5000）。
• 用来匹配工作负载配置中 WorkloadEndpointURL 值的端口。
• 所有 Fabric API 调用均应通过开发网关路由至此端点。

配置工作负荷终结点（用于 Fabric 集成）

与完整的 Microsoft Fabric 开发环境集成时，需要配置工作负荷终结点 URL。 此配置告知开发网关转发 API 请求的位置。

1. 找到或创建工作负荷配置文件（workload-dev-mode.json）：

   • 默认位置为 C:\workload-dev-mode.json。
   • 稍后可以在设置完整的 Fabric 开发环境时创建此文件。

2. 确保 WorkloadEndpointURL 该值与后端端口匹配：

   {
       "WorkloadEndpointURL": "http://localhost:5000",
       // ... other configuration settings
   }
   

有关完整的工作负载配置详细信息，请参阅 扩展性后端入门文档指南。

运行 FastAPI 应用程序

在端口 5000（或所选与配置匹配的端口）上启动 FastAPI 应用程序。

对于 Windows PowerShell：

$env:PYTHONPATH="src"
uvicorn fabric_api.main:app --host 0.0.0.0 --port 5000

对于 Windows 命令提示符：

set PYTHONPATH=src
uvicorn fabric_api.main:app --host 0.0.0.0 --port 5000

对于 macOS 或 Linux：

PYTHONPATH=src uvicorn fabric_api.main:app --host 0.0.0.0 --port 5000

或者，可以从目录中运行命令 src ：

cd src
python -m uvicorn fabric_api.main:app --host 0.0.0.0 --port 5000

验证后端是否可访问

启动应用程序后，验证它是否正常运行：

1. 查看控制台输出。 它应当与以下示例类似：

   INFO:     Uvicorn running on http://0.0.0.0:5000 (Press CTRL+C to quit)
   INFO:     Started reloader process [xxxx]
   INFO:     Started server process [xxxx]
   INFO:     Waiting for application startup.
   INFO:     Application startup complete.
   

2. 测试 API 文档：

   1. 打开浏览器并转到 http://localhost:5000/docs。
   2. 确认 Swagger UI 显示所有可用的终结点。

步骤 7：测试 API

可以使用 FastAPI 提供的 Curl 命令或内置的 Swagger UI 来测试 API。

卷曲

在终端中运行以下命令：

curl -X POST "http://localhost:5000/workspaces/test-workspace/items/TestItemType/test-item-123" \
  -H "Content-Type: application/json" \
  -H "activity-id: test-activity-id" \
  -H "request-id: test-request-123" \
  -H "authorization: SubjectAndAppToken1.0 subjectToken=\"dummy-token\", appToken=\"dummy-app-token\"" \
  -H "x-ms-client-tenant-id: test-tenant-456" \
  -d '{
    "display_name": "Test Item",
    "description": "This is a test item created via curl",
    "creation_payload": {
      "key1": "value1",
      "key2": "value2",
      "nested": {
        "data": "example"
      }
    }
  }'

Swagger UI

FastAPI 自动生成交互式 API 文档，因此可以直接通过浏览器测试终结点：

1. 打开浏览器并转到 http://localhost:5000/docs。

2. 在 ItemLifecycle 部分中，找到 POST 终结点：

   POST /workspaces/{workspaceId}/items/{itemType}/{itemId}
   

3. 选择“试用”按钮。

4. 填写必需的参数：

   • workspaceId: test-workspace
   • itemType: TestItemType
   • itemId: test-item-123
   • activity-id: test-activity-id
   • request-id: test-request-123
   • authorization: SubjectAndAppToken1.0 subjectToken="dummy-token", appToken="dummy-app-token"
   • x-ms-client-tenant-id: test-tenant-456

   对于请求正文，请使用以下代码：

   
   {
       "display_name": "Test Item",
       "description": "This is a test item created via Swagger UI",
       "creation_payload": {
           "key1": "value1",
           "key2": "value2",
           "nested": {
           "data": "example"
           }
       }
   }
   
   

5. 选择“ 执行 ”按钮以发送请求。

   服务器控制台显示类似于以下消息的输出：

   === CREATE ITEM CALLED ===
   Workspace ID: test-workspace
   Item Type: TestItemType
   Item ID: test-item-123
   Display Name: Test Item
   Description: This is a test item created via Swagger UI
   Creation Payload: {'key1': 'value1', 'key2': 'value2', 'nested': {'data': 'example'}}
   ===========================
   

   响应详细信息也直接显示在 Swagger UI 中。

   小窍门

   在开发过程中，使用 Swagger UI 通常更容易、更快，因为它为测试 API 终结点提供了一个用户友好的界面，无需手动创建 Curl 命令。

步骤 8：浏览 API 文档

FastAPI 自动生成交互式 API 文档：

1. 打开浏览器并转到 http://localhost:5000/docs。

2. 在显示的 Swagger UI 上，可以浏览和测试所有终结点。

3. 若要查看创建、获取、更新和删除终结点，请选择该 ItemLifecycle 部分。

下图显示了具有 Fabric API 终结点的 Swagger UI 示例。

步骤 9：实现更高级的功能

前面的步骤提供了有关如何通过 Python 和 FastAPI 实现 ItemLifecycle API 的基本示例。 请记住，本文是一个仅演示核心概念的基础示例。 对于可靠的生产质量后端，通常实现更多功能，例如：

• 创建服务类来处理服务层的业务逻辑、数据库作和其他元素：

  # src/fabric_api/services/storage_service.py
  class StorageService:
      async def create_item(self, workspace_id, item_type, item_id, item_data):
          """
          Store the item in a database or other persistent storage
          """
          # Implementation here
          pass
  
      async def get_item(self, workspace_id, item_type, item_id):
          """
          Retrieve an item from storage
          """
          # Implementation here
          pass
  

• 在控制器中使用依赖项注入：

  # src/fabric_api/impl/item_lifecycle_controller.py
  from fabric_api.services.storage_service import StorageService
  
  class ItemLifecycleController(BaseItemLifecycleApi):
      def __init__(self):
          self.storage_service = StorageService()
  
      async def item_lifecycle_create_item(self, workspaceId, ...):
          # Use the service
          await self.storage_service.create_item(workspaceId, itemType, itemId, create_item_request)
  

• 添加错误处理：

  async def item_lifecycle_create_item(self, ...):
      try:
          # Your implementation
          await self.storage_service.create_item(...)
          return None
      except ValueError as e:
          # Client error
          raise HTTPException(status_code=400, detail=str(e))
      except Exception as e:
          # Server error
          raise HTTPException(status_code=500, detail="Internal server error")
  

下面是可靠后端需要注意的更多事项：

• 其余控制器的实现：例如，实现作业 API 和终结点解析 API。
• 身份验证和授权：通过验证令牌和权限来帮助保护终结点。 有关详细信息，请参阅 后端身份验证和授权概述。
• 持久存储：与数据库或其他存储解决方案集成，实现数据持久性。
• 日志记录和监视：实现全面的日志记录和监视，以跟踪应用程序运行状况和性能。
• 测试：编写单元和集成测试，以帮助确保可靠性和正确性。

结论

现在，你已成功使用 Python 和 FastAPI 设置 Microsoft Fabric 工作负荷 API 后端。 此实现：

• 使用 OpenAPI 生成器工具创建 FastAPI 项目。
• 实现处理 Fabric API 请求所需的控制器。

本文是一个基本示例，演示如何使用 Python 实现 ItemLifecycle API。 为了构建适合生产环境的高质量、可靠且安全的后端，需要执行更多增强功能和注意事项，如 步骤 9 中概述的增强功能和注意事项。

与 Microsoft Fabric 的完全集成需要实现适当的身份验证处理、持久性存储、全面的错误处理和特定于工作负荷的自定义业务逻辑。

相关内容

• Microsoft Fabric 文档
• 工作负载开发工具包文档
• FastAPI 文档
• OpenAPI 生成器文档
• Python asyncio 文档