Rychlý start: Vytvoření bezserverové aplikace pomocí Azure Functions a služby Azure SignalR v Pythonu

Vytvoření projektu funkce Azure Functions

Vytvořte místní projekt funkce Azure Functions.

1. Z příkazového řádku vytvořte adresář pro projekt.
2. Přejděte do adresáře projektu.
3. K inicializaci projektu funkce použijte příkaz Azure Functions func init .

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

Vytvoření funkcí

Po inicializaci projektu je potřeba vytvořit funkce. Tento projekt vyžaduje tři funkce:

• index: Hostuje webovou stránku klienta.
• negotiate: Umožňuje klientovi získat přístupový token.
• broadcast: Používá aktivační událost času k pravidelnému vysílání zpráv všem klientům.

Když spustíte func new příkaz z kořenového adresáře projektu, azure Functions Core Tools připojí kód funkce do function_app.py souboru. Podle potřeby upravíte obsah reklam parametrů tak, že nahradíte výchozí kód kódem aplikace.

Vytvoření indexové funkce

Tuto ukázkovou funkci můžete použít jako šablonu pro vlastní funkce.

Otevřete soubor function_app.py. Tento soubor bude obsahovat vaše funkce. Nejprve upravte soubor tak, aby zahrnoval nezbytné příkazy importu, a definujte globální proměnné, které budeme používat v následujících funkcích.

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

app = func.FunctionApp()

etag = ''
start_count = 0

2. Přidejte funkci index přidáním následujícího kódu.

@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')

Tato funkce hostuje webovou stránku klienta.

Vytvoření funkce negotiate

Přidejte funkci negotiate přidáním následujícího kódu.

@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)

Tato funkce umožňuje klientovi získat přístupový token.

Vytvořte funkci všesměrového vysílání.

Přidejte funkci broadcast přidáním následujícího kódu.

@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) ]
    }))

Tato funkce používá aktivační událost času k pravidelnému vysílání zpráv všem klientům.

Vytvoření projektu funkce Azure Functions

Vytvořte místní projekt funkce Azure Functions.

1. Z příkazového řádku vytvořte adresář pro projekt.
2. Přejděte do adresáře projektu.
3. K inicializaci projektu funkce použijte příkaz Azure Functions func init .

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

Vytvoření funkcí

Po inicializaci projektu je potřeba vytvořit funkce. Tento projekt vyžaduje tři funkce:

• index: Hostuje webovou stránku klienta.
• negotiate: Umožňuje klientovi získat přístupový token.
• broadcast: Používá aktivační událost času k pravidelnému vysílání zpráv všem klientům.

Když spustíte func new příkaz z kořenového adresáře projektu, Azure Functions Core Tools vytvoří výchozí zdrojové soubory funkcí a uloží je do složky pojmenované po funkci. Soubory upravíte podle potřeby a nahradíte výchozí kód kódem aplikace.

Vytvoření indexové funkce

Tuto ukázkovou funkci můžete použít jako šablonu pro vlastní funkce.

1. Spuštěním následujícího příkazu vytvořte index funkci.

func new -n index -t HttpTrigger

1. Upravte index/function.json a nahraďte obsah následujícím kódem JSON:

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

1. Upravte index/_init_.py a nahraďte obsah následujícím kódem:

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')

Vytvoření funkce negotiate

1. Spuštěním následujícího příkazu vytvořte negotiate funkci.

func new -n negotiate -t HttpTrigger

1. Upravte negotiate/function.json a nahraďte obsah následujícím kódem 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. Upravte negotiate/_init_.py a nahraďte obsah následujícím kódem:

import azure.functions as func


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

Vytvořte funkci všesměrového vysílání.

1. Spuštěním následujícího příkazu vytvořte broadcast funkci.

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

1. Upravte všesměrové vysílání/function.json a nahraďte obsah následujícím kódem:

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

1. Upravte broadcast/_init_.py a nahraďte obsah následujícím kódem:

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) ]
    }))