Quickstart: Create a serverless app with Azure Functions and Azure SignalR Service in Python

Create the Azure Function project

Create a local Azure Function project.

1. From a command line, create a directory for your project.
2. Change to the project directory.
3. Use the Azure Functions func init command to initialize your function project.

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

Create the functions

After you initialize a project, you need to create functions. This project requires three functions:

• index: Hosts a web page for a client.
• negotiate: Allows a client to get an access token.
• broadcast: Uses a time trigger to periodically broadcast messages to all clients.

When you run the func new command from the root directory of the project, the Azure Functions Core Tools appends the function code in the function_app.py file. You'll edit the parameters ad content as necessary by replacing the default code with the app code.

Create the index function

You can use this sample function as a template for your own functions.

Open the file function_app.py. This file will contain your functions. First, modify the file to include the neccessary import statements, and define global variables that we will be using in the following functions.

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

app = func.FunctionApp()

etag = ''
start_count = 0

2. Add the function index by adding the following code

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

This function hosts a web page for a client.

Create the negotiate function

Add the function negotiate by adding the following code

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

This function allows a client to get an access token.

Create a broadcast function.

Add the function broadcast by adding the following code

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

This function uses a time trigger to periodically broadcast messages to all clients.

Create the Azure Function project

Create a local Azure Function project.

1. From a command line, create a directory for your project.
2. Change to the project directory.
3. Use the Azure Functions func init command to initialize your function project.

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

Create the functions

After you initialize a project, you need to create functions. This project requires three functions:

• index: Hosts a web page for a client.
• negotiate: Allows a client to get an access token.
• broadcast: Uses a time trigger to periodically broadcast messages to all clients.

When you run the func new command from the root directory of the project, the Azure Functions Core Tools creates default function source files and stores them in a folder named after the function. You'll edit the files as necessary replacing the default code with the app code.

Create the index function

You can use this sample function as a template for your own functions.

1. Run the following command to create the index function.

func new -n index -t HttpTrigger

1. Edit index/function.json and replace the contents with the following json code:

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

1. Edit index/_init_.py and replace the contents with the following code:

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

Create the negotiate function

1. Run the following command to create the negotiate function.

func new -n negotiate -t HttpTrigger

1. Edit negotiate/function.json and replace the contents with the following json code:

{
  "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. Edit negotiate/_init_.py and replace the contents with the following code:

import azure.functions as func


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

Create a broadcast function.

1. Run the following command to create the broadcast function.

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

1. Edit broadcast/function.json and replace the contents with the following code:

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

1. Edit broadcast/_init_.py and replace the contents with the following code:

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