快速入门：采用 Python 通过 Azure Functions 和 Azure SignalR 服务创建无服务器应用

创建 Azure 函数项目

创建本地 Azure 函数项目。

1. 通过命令行创建项目的目录。
2. 更改为项目目录。
3. 使用 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

2. 通过添加以下代码添加函数 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 函数项目。

1. 通过命令行创建项目的目录。
2. 更改为项目目录。
3. 使用 Azure Functions func init 命令初始化函数项目。

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

创建函数

初始化项目后，需要创建函数。 此项目需要三个函数：

• index：托管客户端的网页。
• negotiate：允许客户端获取访问令牌。
• broadcast：使用时间触发器定期将消息广播到所有客户端。

从项目的根目录运行 func new 命令时，Azure Functions Core Tools 会创建默认的函数源文件，并将其存储在以函数命名的文件夹中。 根据需要编辑文件，将默认代码替换为应用代码。

创建索引函数

可将此示例函数用作自己的函数的模板。

1. 运行以下命令来创建 index 函数。

func new -n index -t HttpTrigger

1. 编辑 index/function.json，并将内容替换为以下 json 代码：

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

1. 编辑 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')

创建协商函数

1. 运行以下命令来创建 negotiate 函数。

func new -n negotiate -t HttpTrigger

1. 编辑 negotiate/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"
    }
  ]
}

1. 编辑 negotiate/_init_.py 并将内容替换为以下代码：

import azure.functions as func


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

创建广播函数。

1. 运行以下命令来创建 broadcast 函数。

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

1. 编辑 broadcast/function.json，并将内容替换为以下代码：

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

1. 编辑 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) ]
    }))