时间戳

首页 > 解决方案 > 如何在python中使用定时器触发天蓝色函数v2时向事件发送数据

python-3.x - 如何在python中使用定时器触发天蓝色函数v2时向事件发送数据

问题描述

我正在尝试使用 Azure Functions v2 每 10 秒向事件中心发送一次数据。这是我的function.json代码

    {
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "mytimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "*/10 * * * * *"
    },
    {
      "type": "eventHub",
      "name": "$return",
      "eventHubName": "sqlserverstreaming",
      "connection": "amqps://desrealtimestreaming.servicebus.windows.net",
      "direction": "out"
  }
  ]
}

这是我的init .py代码

    import datetime
import logging
import azure.functions as func


def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()
    print('saran')
    return 'Message created at: {}'.format(utc_timestamp)
    # if mytimer.past_due:
    #     logging.info('The timer is past due!')
    # logging.info('Python timer trigger function ran at %s', utc_timestamp)

我收到以下错误。

 Executed 'Functions.TimerTriggerFUnction' (Failed, Id=fa924331-418f-427f-b672-f525c3ee6b61)
[07-08-2019 09:03:40] System.Private.CoreLib: Exception while executing function: Functions.TimerTriggerFUnction. System.Private.CoreLib: Result: Failure
Exception: FunctionLoadError: cannot load the TimerTriggerFUnction function: Python return annotation "NoneType" does not match binding type "eventHub"
Stack:   File "C:\Users\SivaSakthiVelan\AppData\Roaming\npm\node_modules\azure-functions-core-tools\bin\workers\python\deps\azure\functions_worker\dispatcher.py", line 240, in _handle__function_load_request
    function_id, func, func_request.metadata)
  File "C:\Users\SivaSakthiVelan\AppData\Roaming\npm\node_modules\azure-functions-core-tools\bin\workers\python\deps\azure\functions_worker\functions.py", line 241, in add_function
    f'Python return annotation "{return_pytype.__name__}" '

标签： python-3.xazure-functionsazure-eventhubazure-functions-core-toolsazure-function-async

解决方案

将您的返回类型从更改None为str

def main(mytimer: func.TimerRequest) -> str:

有关完整示例，请参见此处： https ://docs.microsoft.com/en-us/azure/azure-functions/functions-bindings-event-hubs#output---python-example

此外，您的connection属性应该具有包含连接字符串的应用程序设置的名称，而不是连接字符串本身（另请参见示例），如下所示：

  "type": "eventHub",
  "name": "$return",
  "eventHubName": "sqlserverstreaming",
  "connection": "MyConnStringAppSetting",
  "direction": "out"

然后创建一个名为的应用程序设置MyConnStringAppSetting并将您的完整连接字符串放在那里。

推荐阅读