使用 OpenAPI 規格設定 Microsoft 網狀架構工作負載後端 （Swagger）

Microsoft Fabric 工作負載後端是實作 Fabric API 合約的服務，可讓自定義工作負載與 Microsoft Fabric 平臺緊密整合。 此後端會處理工作負載專案的生命週期作業，包括建立、擷取、更新和刪除。

本文會引導您直接從 OpenAPI （Swagger） 定義快速產生 Fabric 工作負載後端。 此 API 優先方法可讓您在將後端邏輯整合到完整的Microsoft網狀架構開發環境中之前，快速建立原型並驗證後端邏輯。 不論您選擇的特定工具或語言為何，這裡所示範的原則都廣泛適用。

本文結束時，您將能夠：

• 根據範例中包含的 Swagger 檔案產生 Fabric 工作負載後端。
• 瞭解 Fabric 工作負載後端的基本結構和組件。
• 使用 Python 和 FastAPI 在本機執行並測試產生的後端。

在本文中，您會從專案生命週期實作下列核心作業。 這些作業會對應至 Fabric API Swagger 檔案中定義的端點。

• 建立專案：初始化新的工作負載專案。
• 取得項目有效負荷：擷取項目配置。
• 更新專案：修改現有的專案。
• 刪除專案：從工作區移除專案。

本文特別示範使用 Python 和 FastAPI 的過程，並搭配 OpenAPI 產生器工具。 不過，OpenAPI 產生器本身支援 許多程式設計語言和架構。 您可以選擇任何符合小組專業知識和專案需求的 OpenAPI 相容程式代碼產生工具或方法，以建立後端基本架構。

先決條件

在開始本文中的步驟之前，請確定您有下列項目。

必要的知識

• 瞭解 Microsoft Fabric 專案生命週期。 讀取 專案生命週期管理。

  本文對於這項理解至關重要。 產生的後端會實作 Fabric 專案的生命週期作業（建立、讀取、更新、刪除），如 專案生命週期檔中所定義。

• Python 和 RESTful API 的基本知識。

• 熟悉Microsoft網狀架構工作負載概念。

必要軟體

• Python 3.8+。 下載 Python。
• Node.js，如果您想要透過 npm 安裝 OpenAPI 產生器 CLI，這是必要的。 下載 Node.js。
• Git，用來複製範例存放庫。 下載 Git。
• 程式代碼編輯器，例如 Visual Studio Code、PyCharm 或您慣用的整合開發環境（IDE）。

Java for OpenAPI 產生器

OpenAPI 產生器 CLI 需要 Java 作為運行時間環境。 您不需要撰寫 Java 程式代碼。 您只需要執行產生器工具。

所需的最低 Java 版本是 Java 8。 我們建議您使用支持的長期支援 （LTS） 版本，例如 Java 17 或 Java 21。

若要安裝 Java：

1. 安裝 Microsoft組建的OpenJDK（值得推薦）。 遵循 安裝 Microsoft Build of 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)
   

如果您已經從另一個廠商安裝 Java（例如 Oracle、Eclipse Temurin 或 Amazon Corretto），且版本為 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

這很重要

如果輸出指向不同的位置（例如您的全系統 Python 安裝），則虛擬環境未正確啟動。 重新檢視 啟動任務 ，並確定您的終端機提示會出現與 (.venv)。

設定 IDE （選擇性 ）

大部分的新式 IDE 會自動偵測 Python 虛擬環境。 不過，您可能需要手動選取 IDE 設定內的解釋器。

範例：Visual Studio Code 設定

1. 在 Visual Studio Code 中開啟您的項目資料夾。

2. 開啟命令選擇區：

   • Windows 或 Linux： Ctrl+Shift+P
   • macOS：Cmd+Shift+P

3. 搜尋並選取 Python: Select Interpreter。

4. 選擇位於虛擬環境中的解釋器：

   • Windows ：.venv\Scripts\python.exe
   • macOS 或 Linux：.venv/bin/python

5. 請在 Visual Studio Code 底部的狀態列上確認您的選擇。 它應該會顯示如下的內容：

   Python 3.x.x ('.venv': venv)
   

6. 開啟新的整合式終端機（終端>新增終端機）。 您的虛擬環境應該會自動啟動，如 (.venv) 提示所示。

針對虛擬環境進行疑難解答

在安裝相依性或執行應用程式之前，請務必確定您的虛擬環境已啟用。 (.venv)終端機中的前置詞會確認啟用狀態。 如果您遇到匯入錯誤或遺失的套件，請執行先前提及的驗證命令，確認您使用正確的 Python 解釋器。

小提示

如果您的 IDE 不會自動偵測您的虛擬環境，或解釋器路徑不符合您的虛擬環境，請嘗試下列解決方案：

• 請確定您已從正確的項目目錄開啟 IDE。
• 重新啟動 IDE，然後再次嘗試選取解釋器。
• 確認您的虛擬環境已在終端機中啟用。

步驟 3：從 OpenAPI 規格產生 FastAPI 專案

使用 OpenAPI 產生器 CLI 從網狀架構 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 執行下列動作：

參數	價值觀	說明	為必填項目	目標	參考文獻
-i	[InputSpecPath] 1	輸入規格： 指定來源 OpenAPI （Swagger） 定義檔的路徑	為必填項目	指向定義所有端點、模型和作業的網狀架構 API 合約	OpenAPI 規格
-g	python-fastapi 2	產生器名稱： 告知工具使用 python-fastapi 產生器來建立伺服器端 Python 程式代碼	為必填項目	決定所產生後端程式代碼的輸出架構和語言	Python FastAPI 產生器 探索所有可用的伺服器產生器
-o	.	輸出目錄： 指示產生器將輸出檔案放在目前目錄中	為必填項目	指定建立產生的項目結構的位置	不適用
--additional-properties	packageName=fabric_api	產生器特定選項： 將產生的程式代碼的 Python 套件名稱設定為 fabric_api	可選	自定義產生的程式代碼結構和命名慣例	產生器選項

¹ 針對 [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 控制器

建立處理網狀架構 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 開發環境。 此步驟對於與網狀架構開發閘道的適當整合至關重要。

瞭解埠設定

當您開發 Microsoft Fabric 負載時，開發網關會將 API 要求路由至您的後端。 這個設定需要：

• 您的後端會在特定埠上執行（預設值：5000）。
• 與工作負載組態中的 WorkloadEndpointURL 值相符的端口。
• 所有應該透過開發閘道進行路由的 Fabric API 呼叫將傳送至此端點。

設定工作負載端點 （適用於網狀架構整合）

當您與完整Microsoft網狀架構開發環境整合時，您必須設定工作負載端點 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

這很重要

設定 PYTHONPATH 對於 Python 正確尋找模組至關重要。 此環境變數只會影響當前的終端模擬器會話。

或者，您可以從 目錄執行 命令 src ：

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

備註

埠 5000 通常用於 Microsoft Fabric 工作負載開發範例中的默認。 如果您需要使用不同的連接埠：

1. 變更您指令 --port 中的 uvicorn 值（例如，--port 5001）。
2. 若要比對這個新埠，請更新 WorkloadEndpointURL 檔案 workload-dev-mode.json 中的值（例如 ， "http://localhost:5001"）。

請確定系統中的另一個應用程式尚未使用您選擇的埠。

確認您的後端系統可存取

啟動應用程式之後，請確認它正在正確執行：

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

您可以使用 Curl 命令或 FastAPI 提供的內建 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 段。

下圖顯示具有網狀架構 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網狀架構工作負載 API 後端。 此實作：

• 使用 OpenAPI 產生器工具來建立 FastAPI 專案。
• 實作處理網狀架構 API 要求的必要控制器。

本文是示範如何使用 Python 實作 API ItemLifecycle 的基本範例。 需要更多增強功能和考慮，例如 步驟 9：實作更進階的功能，才能建置適合生產環境的高品質、健全且安全的後端。

與 Microsoft Fabric 的完整整合需要實作適當的驗證處理、持續性記憶體、完整的錯誤處理，以及您工作負載特有的自定義商業規則。

相關內容

• Microsoft網狀架構檔
• 工作負載開發工具包檔
• FastAPI 文件
• OpenAPI 產生器檔
• Python asyncio 文件