開始使用 Python 中的轉送混合式連線 WebSocket 要求

在本快速入門中，您會使用 WebSocket 通訊協定建立 Python 傳送者和接收者應用程式，以傳送和接收訊息。 應用程式將使用 Azure 轉送的混合式連線功能。 若要對 Azure 轉送有整體上的了解，請參閱 Azure 轉送。

在本快速入門中，您會執行下列步驟：

1. 使用 Azure 入口網站建立轉送命名空間。
2. 使用 Azure 入口網站，在該命名空間中建立混合式連線。
3. 產生config.json屬性檔案以儲存連線詳細數據
4. 開發協助程式函式的 relaylib.py（SAS 令牌、URL）
5. 撰寫伺服器 （接聽程式） 文稿以接收訊息。
6. 撰寫用戶端 （sender） 腳本以傳送訊息。
7. 執行伺服器 （接聽程式） 文稿，並選擇性地執行用戶端 （sender） 文稿。

必要條件

• Python 確定您正在執行 Python 3.10+
• Azure 訂用帳戶。 如果您沒有 Azure 訂用帳戶，請在開始前建立免費帳戶。

使用 Azure 入口網站建立命名空間

1. 登入 Azure 入口網站。

2. 選取左側功能表上的 [所有服務]。 選取 [整合]，並搜尋「轉送」，再將滑鼠移至 [轉送] 上方，然後選取 [建立]。

   

3. 在 [建立命名空間] 頁面上，遵循下列步驟：

   1. 選擇要在其中建立命名空間的 Azure 訂用帳戶。

   2. 針對資源群組，選擇將放置命名空間的現有資源群組，或是建立新的資源群組。

   3. 輸入轉送命名空間的名稱。

   4. 選取要用來裝載命名空間的區域。

   5. 選取頁面底部的 [檢閱 + 建立] 。

      

   6. 在 [檢閱 + 建立] 頁面上，選取 [建立]。

   7. 幾分鐘後，您會看到命名空間的 [轉送] 頁面。

      

取得管理認證

1. 在 [轉送] 頁面上，選取左側功能表上的 [共用存取原則]。

2. 在 [共用存取原則] 頁面上，選取 [RootManageSharedAccessKey]。

3. 在 [SAS 原則: RootManageSharedAccessKey] 下，選取 [主要連接字串] 旁的 [複製] 按鈕。 此動作會將連接字串複製到剪貼簿以供稍後使用。 將此值貼到記事本或一些其他暫存位置。

4. 重複前一個步驟，複製 [主要金鑰] 的值並貼到暫存位置以供稍後使用。

   

使用 Azure 入口網站建立混合式連線

在命名空間的 [轉送] 頁面上，遵循下列步驟來建立混合式連線。

1. 在左側功能表的 [實體] 下，選取 [混合式連線]，然後選取 [+ 混合式連線]。

   

2. 在 [建立混合式連線] 頁面上，輸入混合式連線的名稱，然後選取 [建立]。

   

開發協助程式函式

建立 Python 腳本

此腳本會針對使用 Azure 轉送混合式連線的應用程式提供協助程式函式。 這些函式可能會協助產生SAS令牌和建立WebSocket線上等工作，以進行安全通訊。

相依性

在產生協助程式函式文稿之前，請先使用 pip 安裝下列 Python 連結庫：base64、、hashlibhmac、、 timemathurllib

您可以使用下列命令來安裝這些連結庫：

pip install <package name>

撰寫協助程式函式腳本

您的 relaylib.py 檔案看起來應該像下面這樣：

import base64
import hashlib
import hmac
import math
import time
import urllib

# Function which generates the HMAC-SHA256 of a given message
def hmac_sha256(key, msg):
   hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
   return hash_obj.digest()

# Function to create a WebSocket URL for listening for a server application
def createListenUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
   if token is not None:
       url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the url for the client application
def createSendUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
   if token is not None:
	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the Service Bus SAS token. 
def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
   uri = "http://" + serviceNamespace + "/" + entityPath
   encodedResourceUri = urllib.parse.quote(uri, safe = '')

   # Define the token validity period in seconds (48 hours in this case)   
   tokenValidTimeInSeconds = 60 * 60 * 48 
   unixSeconds = math.floor(time.time())
   expiryInSeconds = unixSeconds + tokenValidTimeInSeconds

   # Create the plain signature string by combining the encoded URI and the expiry time
   plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)

   # Encode the SAS key and the plain signature as bytes
   sasKeyBytes = sasKey.encode("utf-8")
   plainSignatureBytes = plainSignature.encode("utf-8")
   hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
   base64HashValue = base64.b64encode(hashBytes)

    # Construct the SAS token string
   token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
   return token

建立伺服器應用程式 (接聽程式)

若要接聽和接收來自轉寄的訊息，請撰寫 Python WebSocket 伺服器腳本。

建立 Python 腳本

如果您在建立轉送時停用 [需要用戶端授權] 選項，您可以使用任何瀏覽器將要求傳送至混合式連線 URL。 若要存取受保護的端點，您必須建立並傳遞 SAS 令牌，如下所示。

以下是簡單的 Python 腳本，示範如何使用 WebSocket 將要求傳送至混合式連線 URL 與 SAS 令牌。

相依性

1. 在執行伺服器應用程式之前，請先使用 pip 安裝下列 Python 連結庫

   asyncio、 、 json、 loggingwebsockets

   您可以使用下列命令來安裝這些連結庫：

    pip install <package name>
   

2. 產生檔案 config.json 以儲存連線詳細數據

    {
   "namespace": "HYBRID_CONNECTION_NAMESPACE",
   "path": "HYBRID_CONNECTION_ENTITY_NAME",
   "keyrule": "SHARED_ACCESS_KEY_NAME",
   "key": "SHARED_ACCESS_PRIMARY_KEY"
    }
   

   將方括號中的預留位置取代為您在建立混合式連線時所取得的值。

   • namespace - 轉送命名空間。 務必使用完整命名空間名稱；例如，{namespace}.servicebus.windows.net。
   • path - 混合式連線的名稱。
   • keyrule - 預設為共用存取原則金鑰 RootManageSharedAccessKey 的名稱。
   • key - 您稍早儲存之命名空間的主鍵。

3. 產生協助程式函式的協助程式函式檔案

   下列檔案用來作為 relaylib.py WebSocket URL 產生和 SAS 令牌的協助程式函式

   建立 Python 腳本

   此腳本會針對使用 Azure 轉送混合式連線的應用程式提供協助程式函式。 這些函式可能會協助產生SAS令牌和建立WebSocket線上等工作，以進行安全通訊。

   相依性

   在產生協助程式函式文稿之前，請先使用 pip 安裝下列 Python 連結庫：base64、、hashlibhmac、、 timemathurllib

   您可以使用下列命令來安裝這些連結庫：

   pip install <package name>
   

   撰寫協助程式函式腳本

   您的 relaylib.py 檔案看起來應該像下面這樣：

   import base64
   import hashlib
   import hmac
   import math
   import time
   import urllib
   
   # Function which generates the HMAC-SHA256 of a given message
   def hmac_sha256(key, msg):
      hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
      return hash_obj.digest()
   
   # Function to create a WebSocket URL for listening for a server application
   def createListenUrl(serviceNamespace, entityPath, token = None):
      url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
      if token is not None:
          url = url + '&sb-hc-token=' + urllib.parse.quote(token)
      return url
   
   # Function which creates the url for the client application
   def createSendUrl(serviceNamespace, entityPath, token = None):
      url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
      if token is not None:
   	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
      return url
   
   # Function which creates the Service Bus SAS token. 
   def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
      uri = "http://" + serviceNamespace + "/" + entityPath
      encodedResourceUri = urllib.parse.quote(uri, safe = '')
   
      # Define the token validity period in seconds (48 hours in this case)   
      tokenValidTimeInSeconds = 60 * 60 * 48 
      unixSeconds = math.floor(time.time())
      expiryInSeconds = unixSeconds + tokenValidTimeInSeconds
   
      # Create the plain signature string by combining the encoded URI and the expiry time
      plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)
   
      # Encode the SAS key and the plain signature as bytes
      sasKeyBytes = sasKey.encode("utf-8")
      plainSignatureBytes = plainSignature.encode("utf-8")
      hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
      base64HashValue = base64.b64encode(hashBytes)
   
       # Construct the SAS token string
      token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
      return token
   

撰寫一些程式碼來傳送訊息

1. 請確定您的相依性， config.json 且 relaylib.py 可在您的路徑中使用

2. 您的 listener.py 檔案看起來應該像下面這樣：

    import asyncio
    import json
    import logging
    import relaylib
    import websockets
   
    async def run_application(config):
        serviceNamespace = config["namespace"]
        entityPath = config["path"]
        sasKeyName = config["keyrule"]
        sasKey = config["key"]
        serviceNamespace += ".servicebus.windows.net"
        # Configure logging
        logging.basicConfig(level=logging.INFO)  # Enable DEBUG/INFO logging as appropriate
   
        try:
            logging.debug("Generating SAS Token for: %s", serviceNamespace)
    	token = relaylib.createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey)
    	logging.debug("Generating WebSocket URI")
    	wssUri = relaylib.createListenUrl(serviceNamespace, entityPath, token)
    	async with websockets.connect(wssUri) as websocket:
    	    logging.info("Listening for messages on Azure Relay WebSocket...")
    	    while True:
    		message = await websocket.recv()
    		logging.info("Received message: %s", message)
    	    except KeyboardInterrupt:
    		logging.info("Exiting listener.")
   
    if __name__ == "__main__":
        # Load configuration from JSON file
        with open("config.json") as config_file:
           config = json.load(config_file)
   
        asyncio.run(run_application(config))
   

建立用戶端應用程式 (傳送者)

若要將訊息傳送至轉送，您可以使用任何 HTTP 或 WebSocket 用戶端，包含的範例是 Python 實作。

建立 Python 腳本

如果您在建立轉送時停用 [需要用戶端授權] 選項，您可以使用任何瀏覽器將要求傳送至混合式連線 URL。 若要存取受保護的端點，您必須建立並傳遞 SAS 令牌，如下所示。

以下是簡單的 Python 腳本，示範如何使用 WebSocket 將要求傳送至混合式連線 URL 與 SAS 令牌。

相依性

1. 在執行用戶端應用程式之前，請先使用 pip 安裝下列 Python 連結庫

   asyncio、 、 json、 loggingwebsockets

   您可以使用下列命令來安裝這些連結庫：

    pip install <package name>
   

2. 產生檔案 config.json 以儲存連線詳細數據

    {
   "namespace": "HYBRID_CONNECTION_NAMESPACE",
   "path": "HYBRID_CONNECTION_ENTITY_NAME",
   "keyrule": "SHARED_ACCESS_KEY_NAME",
   "key": "SHARED_ACCESS_PRIMARY_KEY"
   }
   

   將方括號中的預留位置取代為您在建立混合式連線時所取得的值。

   • namespace - 轉送命名空間。 務必使用完整命名空間名稱；例如，{namespace}.servicebus.windows.net。
   • path - 混合式連線的名稱。
   • keyrule - 預設為共用存取原則金鑰 RootManageSharedAccessKey 的名稱。
   • key - 您稍早儲存之命名空間的主鍵。

3. 產生協助程式函式的協助程式函式檔案

   下列檔案用來作為 relaylib.py WebSocket URL 產生和 SAS 令牌的協助程式函式

   建立 Python 腳本

   此腳本會針對使用 Azure 轉送混合式連線的應用程式提供協助程式函式。 這些函式可能會協助產生SAS令牌和建立WebSocket線上等工作，以進行安全通訊。

   相依性

   在產生協助程式函式文稿之前，請先使用 pip 安裝下列 Python 連結庫：base64、、hashlibhmac、、 timemathurllib

   您可以使用下列命令來安裝這些連結庫：

   pip install <package name>
   

   撰寫協助程式函式腳本

   您的 relaylib.py 檔案看起來應該像下面這樣：

   import base64
   import hashlib
   import hmac
   import math
   import time
   import urllib
   
   # Function which generates the HMAC-SHA256 of a given message
   def hmac_sha256(key, msg):
      hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
      return hash_obj.digest()
   
   # Function to create a WebSocket URL for listening for a server application
   def createListenUrl(serviceNamespace, entityPath, token = None):
      url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
      if token is not None:
          url = url + '&sb-hc-token=' + urllib.parse.quote(token)
      return url
   
   # Function which creates the url for the client application
   def createSendUrl(serviceNamespace, entityPath, token = None):
      url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
      if token is not None:
   	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
      return url
   
   # Function which creates the Service Bus SAS token. 
   def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
      uri = "http://" + serviceNamespace + "/" + entityPath
      encodedResourceUri = urllib.parse.quote(uri, safe = '')
   
      # Define the token validity period in seconds (48 hours in this case)   
      tokenValidTimeInSeconds = 60 * 60 * 48 
      unixSeconds = math.floor(time.time())
      expiryInSeconds = unixSeconds + tokenValidTimeInSeconds
   
      # Create the plain signature string by combining the encoded URI and the expiry time
      plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)
   
      # Encode the SAS key and the plain signature as bytes
      sasKeyBytes = sasKey.encode("utf-8")
      plainSignatureBytes = plainSignature.encode("utf-8")
      hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
      base64HashValue = base64.b64encode(hashBytes)
   
       # Construct the SAS token string
      token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
      return token
   

撰寫一些程式碼來傳送訊息

1. 請確定您的相依性， config.json 且 relaylib.py 可在您的路徑中使用

2. 您的 sender.py 檔案看起來應該像下面這樣：

    import asyncio
    import json
    import logging
    import relaylib
    import websockets
   
    async def run_application(message, config):
    	service_namespace = config["namespace"]
    	entity_path = config["path"]
    	sas_key_name = config["keyrule"]
    	sas_key = config["key"]
    	service_namespace += ".servicebus.windows.net"
   
    	# Configure logging
    	logging.basicConfig(level=logging.DEBUG)  # Enable debug logging
   
    	token = relaylib.createSasToken(service_namespace, entity_path, sas_key_name, sas_key)
    	logging.debug("Token: %s", token)
    	wss_uri = relaylib.createListenUrl(service_namespace, entity_path, token)
    	logging.debug("WssURI: %s", wss_uri)
   
    	try:
    		async with websockets.connect(wss_uri) as websocket:
    			logging.info("Sending message to Azure Relay WebSocket...")
    			await websocket.send(json.dumps({"message": message}))
    			logging.info("Message sent: %s", message)
    	except Exception as e:
    		logging.error("An error occurred: %s", str(e))
   
    if __name__ == "__main__":
    	# Load configuration from JSON file
    	with open("config.json") as config_file:
    		config = json.load(config_file)
   
    	asyncio.run(run_application("This is a message to Azure Relay Hybrid Connections!", config))
   

執行應用程式

1. 執行伺服器應用程式：從命令提示字元輸入 python3 listener.py。
2. 執行用戶端應用程式：從命令提示字元輸入 python3 sender.py。

恭喜您，您已使用 Python 建立端對端混合式連線應用程式！

下一步

在本快速入門中，您已建立使用 WebSocket 來傳送和接收訊息的 Python 用戶端和伺服器應用程式。 Azure 轉送的混合式連線功能也支援使用 HTTP 來傳送和接收訊息。 若要了解如何搭配使用 HTTP 和 Azure 轉送混合式連線，請參閱 HTTP 快速入門。

在本快速入門中，您已使用 Python 來建立客戶端和伺服器應用程式。 若要瞭解如何使用 .NET Framework 撰寫用戶端和伺服器應用程式，請參閱 .NET HTTP 快速入門 或 .NET HTTP 快速入門。