在 Python 中的 Azure Functions 中。将消息发布到 Azure 服务总线时设置 SessionID

Question

我用 Python 起草了一个 Azure 函数，模型 V2。

我已成功向主题发布消息。

现在我想为已发布的消息设置SessionID，但不知道如何操作。如何发布带有 SessionId 的消息？

Azure 无法识别我在“my_json_string”变量中使用 JSON 的尝试。

import logging
import azure.functions as func
from datetime import datetime
import json


app = func.FunctionApp()

# vs output into queue for python
# https://learn.microsoft.com/en-us/azure/azure-functions/functions-bindings-service-bus-output?tabs=python-v2%2Cisolated-process%2Cnodejs-v4%2Cextensionv5&pivots=programming-language-python


@app.route(route="http_trigger_topic", auth_level=func.AuthLevel.ANONYMOUS)
@app.service_bus_topic_output(arg_name="message",
                              connection="ServiceBusConnection",
                              topic_name="alfdevapi6topic")
def http_trigger_topic(req: func.HttpRequest, message: func.Out[str]) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    myMessage = "Hi alf this is my message via the queue to you."
    logging.info(myMessage)
    
    input_msg = req.params.get('message')

    now = datetime.now()
    print("now =", now)

    dt_string = now.strftime("%d/%m/%Y %H:%M:%S")

    my_json_string = f"""
    {{
        "body": "{myMessage}",
        "customProperties": {{
            "messagenumber": 0,
            "timePublish": "{now}",
        }},
        "brokerProperties": {{
            "SessionId": "1"
        }}
    }}
    """


    message.set(my_json_string)


    return func.HttpResponse(
        "This function should process queue messages.",
        status_code=200
    )

Answer 1

我确实同意@Skip的观点，即

func.Out

无法发送带有会话ID的消息。

但是您也可以使用 SDK 使用以下代码：

import azure.functions as func
import logging
from azure.servicebus import ServiceBusClient, ServiceBusMessage

app = func.FunctionApp()
@app.service_bus_topic_trigger(arg_name="azservicebus", subscription_name="mysubscription", topic_name="mysbtopic",
                               connection="sbfiltersid_SERVICEBUS") 
def servicebus_topic_trigger(azservicebus: func.ServiceBusMessage):
    logging.info('Python ServiceBus Topic trigger processed a message: %s',
                azservicebus.get_body().decode('utf-8'))
    rith_con = "Endpoint=sb://sbfiltersid.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=MT5eFCEndpoint=sb://sbfiltersid.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=MT5eFCUG+Ng="
    rith_qn = "rithwik"
    servicebus_client = ServiceBusClient.from_connection_string(rith_con)
    rith_sen = servicebus_client.get_queue_sender(rith_qn)
    rith_msg = ServiceBusMessage("Hello Rithwik Bojja")
    rith_msg.session_id = "008"
    with rith_sen:
        rith_sen.send_messages(rith_msg)
    servicebus_client.close()

输出：