快速入門：在 Python 中使用 Azure Functions 和 Azure SignalR Service 建立無伺服器應用程式

發行項
04/27/2024

使用 Azure Functions 和 Python 開始使用 Azure SignalR Service，以建置無伺服器應用程式，以將訊息廣播至用戶端。您將在本機環境中執行函式，連線到雲端中的 Azure SignalR Service 實例。完成本快速入門會在您的 Azure 帳戶中產生幾美分或更少成本。

注意

您可以從 GitHub 取得本文中的程式代碼。

必要條件

本快速入門可以在 macOS、Windows 或 Linux 上執行。您將需要下列項目：

必要條件	描述
Azure 訂閱	如果您沒有 Azure 訂用帳戶，請建立 Azure 免費帳戶
程式碼編輯器	您將需要程式代碼編輯器，例如 Visual Studio Code。
Azure Functions Core Tools	需要 2.7.1505 版或更高版本，才能在本機執行 Python Azure 函式應用程式。
Python 3.7+	Azure Functions 需要 Python 3.7+。請參閱支援的 Python 版本。
Azurite	SignalR 系結需要 Azure 儲存體。當函式在本機執行時，您可以使用本機記憶體模擬器。
Azure CLI	您可以選擇性地使用 Azure CLI 來建立 Azure SignalR Service 實例。

建立 Azure SignalR 服務實例

在本節中，您會建立要用於應用程式的基本 Azure SignalR 實例。下列步驟會使用 Azure 入口網站來建立新的實例，但您也可以使用 Azure CLI。如需詳細資訊，請參閱 Azure SignalR Service CLI 參考中的 az signalr create 命令。

登入 Azure 入口網站。
在頁面左上角，選取 [+ 建立資源]。
在 [建立資源] 頁面上的 [搜尋服務和市集] 文本框中，輸入 signalr，然後從列表中選取 SignalR 服務。
在 [ SignalR 服務] 頁面上，選取 [ 建立]。
在 [ 基本] 索引 標籤上，輸入新 SignalR Service 實例的基本資訊。輸入下列值：

欄位	建議的值	描述
訂用帳戶	選擇您的訂用帳戶	選取您想要用來建立新 SignalR Service 實例的訂用帳戶。
資源群組	建立名為 SignalRTestResources 的資源群組	選取或建立 SignalR 資源的資源群組。為本教學課程建立新的資源群組，而不是使用現有的資源群組會很有用。若要在完成本教學課程之後釋放資源，請刪除資源群組。刪除資源群組也會刪除屬於群組的所有資源。此動作無法復原。刪除資源群組之前，請確定它不包含您想要保留的資源。如需詳細資訊，請參閱使用資源群組來管理您的 Azure 資源。
資源名稱	testsignalr	輸入要用於 SignalR 資源的唯一資源名稱。如果您的區域中已經採用testignalr ，請新增數位或字元，直到名稱是唯一的。名稱必須是 1 到 63 個字元的字串，且只包含數位、字母和連字元（`-`）字元。名稱不能以連字元字元開頭或結尾，且連續連字元字元無效。
區域	選擇您的區域	為新的 SignalR 服務實例選取適當的區域。 Azure SignalR Service 目前無法在所有區域中使用。如需詳細資訊，請參閱 Azure SignalR 服務區域可用性
定價層	選取 [變更]，然後選擇 [免費]（僅限開發/測試）。選擇 [ 選取 ] 以確認您選擇的定價層。	Azure SignalR Service 有三個定價層：免費、標準和進階版。除非必要條件另有說明，否則教學課程會使用免費層。如需階層與定價之間功能差異的詳細資訊，請參閱 Azure SignalR 服務定價
服務模式	選擇適當的服務模式	當您在 Web 應用程式中裝載 SignalR 中樞邏輯，並使用 SignalR 服務作為 Proxy 時，請使用預設值。當您使用無伺服器技術，例如 Azure Functions 來裝載 SignalR 中樞邏輯時，請使用無伺服器。傳統模式僅供回溯相容性使用，不建議使用。如需詳細資訊，請參閱 Azure SignalR Service 中的服務模式。

您不需要變更 SignalR 教學課程之 [網络 和標籤 ] 索引標籤上的設定。

選取 [基本] 索引標籤底部的 [檢閱 + 建立] 按鈕。
在 [ 檢閱 + 建立] 索引卷標上，檢閱值，然後選取 [ 建立]。部署需要一些時間才能完成。
部署完成時，選取 [移至資源 ] 按鈕。
在 [SignalR 資源] 頁面上，從左側的功能表中，選取 [設定]。
複製主鍵的連線字串。在本教學課程稍後，您需要此連接字串來設定您的應用程式。

建立 Azure 函式專案

建立本機 Azure 函式專案。

從命令行建立項目的目錄。
變更為專案目錄。
使用 Azure Functions func init 命令來初始化函式專案。

# Initialize a function project
func init --worker-runtime python

建立函式

初始化項目之後，您必須建立函式。此專案需要三個函式：

index：裝載用戶端的網頁。
negotiate：允許用戶端取得存取令牌。
broadcast：使用時間觸發程式定期將訊息廣播到所有用戶端。

當您從專案的根目錄執行 func new 命令時，Azure Functions Core Tools 會在檔案中 function_app.py 附加函式程式代碼。您將視需要編輯參數廣告內容，方法是將預設程式代碼取代為應用程式程式代碼。

建立索引函式

您可以使用此範例函式作為您自己的函式範本。

開啟 function_app.py 檔案。此檔案將包含您的函式。首先，修改檔案以包含必要的匯入語句，並定義我們將使用於下列函式中的全域變數。

import azure.functions as func
import os
import requests
import json 

app = func.FunctionApp()

etag = ''
start_count = 0

新增下列程式代碼以新增函 index 式

@app.route(route="index", auth_level=func.AuthLevel.ANONYMOUS)
def index(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

此函式會裝載用戶端的網頁。

建立交涉函式

新增下列程式代碼以新增函 negotiate 式

@app.route(route="negotiate", auth_level=func.AuthLevel.ANONYMOUS, methods=["POST"])
@app.generic_input_binding(arg_name="connectionInfo", type="signalRConnectionInfo", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def negotiate(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

此函式可讓用戶端取得存取令牌。

建立廣播函式。

新增下列程式代碼以新增函 broadcast 式

@app.timer_trigger(schedule="*/1 * * * *", arg_name="myTimer",
              run_on_startup=False,
              use_monitor=False)
@app.generic_output_binding(arg_name="signalRMessages", type="signalR", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def broadcast(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-functions-python-worker', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://api.github.com/repos/azure/azure-functions-python-worker is: ' + str(start_count) ]
    }))

此函式會使用時間觸發程式定期將訊息廣播到所有用戶端。

建立 Azure 函式專案

建立本機 Azure 函式專案。

從命令行建立項目的目錄。
變更為專案目錄。
使用 Azure Functions func init 命令來初始化函式專案。

# Initialize a function project
func init --worker-runtime python --model v1

建立函式

初始化項目之後，您必須建立函式。此專案需要三個函式：

index：裝載用戶端的網頁。
negotiate：允許用戶端取得存取令牌。
broadcast：使用時間觸發程式定期將訊息廣播到所有用戶端。

當您從專案的根目錄執行 func new 命令時，Azure Functions Core Tools 會建立預設函式來源檔案，並將其儲存在以函式命名的資料夾中。您將視需要編輯檔案，並將預設程式代碼取代為應用程式程式碼。

建立索引函式

您可以使用此範例函式作為您自己的函式範本。

執行下列命令以建立函 index 式。

func new -n index -t HttpTrigger

編輯 索引/function.json ，並以下列 json 程式代碼取代內容：

{
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

編輯 index/_init_.py 並以下列程式代碼取代內容：

import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/../content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

建立交涉函式

執行下列命令以建立函 negotiate 式。

func new -n negotiate -t HttpTrigger

編輯 交涉/function.json ，並以下列 json 程式代碼取代內容：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "type": "signalRConnectionInfo",
      "name": "connectionInfo",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "in"
    }
  ]
}

編輯 negotiate/_init_.py 並以下列程式代碼取代內容：

import azure.functions as func


def main(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

建立廣播函式。

執行下列命令以建立函 broadcast 式。

func new -n broadcast -t TimerTrigger
# install requests
pip install requests

編輯 廣播/function.json ，並以下列程式代碼取代內容：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "*/5 * * * * *"
    },
    {
      "type": "signalR",
      "name": "signalRMessages",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "out"
    }
  ]
}

編輯 broadcast/_init_.py 並以下列程式代碼取代內容：

import requests
import json

import azure.functions as func

etag = ''
start_count = 0

def main(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-signalr', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://github.com/Azure/azure-signalr is: ' + str(start_count) ]
    }))

建立index.html檔案

此應用程式的用戶端介面是網頁。函 index 式會從 content/index.html 檔案讀取 HTML 內容。

在專案根資料夾中建立名為 content 的資料夾。
建立檔案 content/index.html。
將下列內容複製到 content/index.html 檔案並加以儲存：

<html>

<body>
  <h1>Azure SignalR Serverless Sample</h1>
  <div id="messages"></div>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/microsoft-signalr/3.1.7/signalr.min.js"></script>
  <script>
    let messages = document.querySelector('#messages');
    const apiBaseUrl = window.location.origin;
    const connection = new signalR.HubConnectionBuilder()
        .withUrl(apiBaseUrl + '/api')
        .configureLogging(signalR.LogLevel.Information)
        .build();
      connection.on('newMessage', (message) => {
        document.getElementById("messages").innerHTML = message;
      });

      connection.start()
        .catch(console.error);
  </script>
</body>

</html>

將 SignalR Service 連接字串新增至函式應用程式設定

最後一個步驟是在 Azure 函式應用程式設定中設定 SignalR Service 連接字串。

在 Azure 入口網站中，移至您稍早部署的 SignalR 實例。
選取 [金鑰] 以檢視 SignalR Service 實例的連接字串。

複製主要連接字串，然後執行命令：

func settings add AzureSignalRConnectionString "<signalr-connection-string>"

在本機執行 Azure 函式應用程式

啟動 Azurite 記憶體模擬器：

azurite

在本機環境中執行 Azure 函式應用程式：

func start

注意

如果您看到在 Blob 記憶體上顯示讀取錯誤的錯誤，請確定 local.settings.json 檔案中的 'AzureWebJobs 儲存體' 設定設為 UseDevelopmentStorage=true。

在 Azure 函式在本機執行之後，請移至 http://localhost:7071/api/index。此頁面會顯示 GitHub Azure/azure-signalr 存放庫目前的星號計數。當您在 GitHub 中將存放庫星號或取消星號時，每隔幾秒鐘就會看到重新整理的計數。

清除資源

如果您不打算繼續使用此應用程式，請使用下列步驟刪除本快速入門所建立的所有資源，因此不會產生任何費用：

在 Azure 入口網站中選取最左側的 [資源群組]，然後選取您所建立的資源群組。或者，您可以使用搜尋方塊依其名稱尋找資源群組。
在開啟的視窗中，選取資源群組，然後按兩下 [ 刪除資源群組]。
在新視窗中輸入要刪除之資源群組的名稱，然後按一下 [刪除]。

有問題嗎？請嘗試疑難解答指南，或讓我們知道。

下一步

在本快速入門中，您已在本機建置並執行即時無伺服器應用程式。接下來，深入瞭解如何使用用戶端與 Azure 函式與 SignalR Service 之間的雙向通訊。

Azure Functions 的 SignalR 服務系結

無伺服器中的雙向通訊

使用 VS Code 部署 Azure Functions

快速入門：在 Python 中使用 Azure Functions 和 Azure SignalR Service 建立無伺服器應用程式

必要條件

建立 Azure SignalR 服務實例

建立 Azure 函式專案

建立函式

建立索引函式

建立交涉函式

建立廣播函式。

建立 Azure 函式專案

建立函式

建立索引函式

建立交涉函式

建立廣播函式。

建立index.html檔案

將 SignalR Service 連接字串 新增至函式應用程式設定

在本機執行 Azure 函式應用程式

清除資源

下一步

其他資源

將 SignalR Service 連接字串新增至函式應用程式設定