Azure Functions Python 开发人员指南

Azure Functions 要求函数是 Python 脚本中处理输入并生成输出的无状态方法。 默认情况下，运行时预期该方法在 __init__.py 文件中作为名为 main() 的全局方法实现。 还可以指定替代入口点。

来自触发器和绑定的数据使用在 function.json 文件中定义的 name 属性，通过方法特性绑定到函数。 例如，以下 function.json 文件描述一个由名为 req 的 HTTP 请求触发的简单函数：

{
    "scriptFile": "__init__.py",
    "bindings": [
        {
            "authLevel": "function",
            "type": "httpTrigger",
            "direction": "in",
            "name": "req",
            "methods": [
                "get",
                "post"
            ]
        },
        {
            "type": "http",
            "direction": "out",
            "name": "$return"
        }
    ]
}

根据这一定义，包含函数代码的 __init__.py 文件可能类似于以下示例：

def main(req):
    user = req.params.get('user')
    return f'Hello, {user}!'

还可以使用 Python 类型注释在函数中显式声明属性类型和返回类型。 此操作有助于使用许多 Python 代码编辑器提供的 IntelliSense 和自动完成功能。

import azure.functions

def main(req: azure.functions.HttpRequest) -> str:
    user = req.params.get('user')
    return f'Hello, {user}!'

使用 azure.functions.* 包中附带的 Python 注释将输入和输出绑定到方法。

Azure Functions 要求函数是 Python 脚本中处理输入并生成输出的无状态方法。 默认情况下，运行时期望方法在 function_app.py 文件中作为全局方法实现。

可以使用基于修饰器的方法声明触发器和绑定，并函数中使用它们。 它们在与函数相同的文件 function_app.py 中定义。 例如，以下 function_app.py 文件表示 HTTP 请求的函数触发器。

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req):
    user = req.params.get("user")
    return f"Hello, {user}!"

还可以使用 Python 类型注释在函数中显式声明属性类型和返回类型。 此操作有助于使用许多 Python 代码编辑器提供的 IntelliSense 和自动完成功能。

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> str:
    user = req.params.get("user")
    return f"Hello, {user}!"

若要了解 v2 模型的已知限制及其解决方法，请参阅排查 Azure Functions 中的 Python 错误。

可以选择在 function.json 文件中指定 scriptFile 和 entryPoint 属性来更改函数的默认行为。 例如，下面的 function.json 指示运行时使用 main.py 文件中的 customentry() 方法作为 Azure 函数的入口点。

{
  "scriptFile": "main.py",
  "entryPoint": "customentry",
  "bindings": [
      ...
  ]
}

入口点只能在 function_app.py 文件中。 但是，可以使用蓝图或通过导入在 function_app.py 中引用项目中的函数。

Python 函数项目的建议文件夹结构如以下示例所示：

 <project_root>/
 | - .venv/
 | - .vscode/
 | - my_first_function/
 | | - __init__.py
 | | - function.json
 | | - example.py
 | - my_second_function/
 | | - __init__.py
 | | - function.json
 | - shared_code/
 | | - __init__.py
 | | - my_first_helper_function.py
 | | - my_second_helper_function.py
 | - tests/
 | | - test_my_second_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

主项目文件夹 <project_root> 可以包含以下文件：

• local.settings.json：用于在本地运行时存储应用设置和连接字符串。 此文件不会被发布到 Azure。 若要了解详细信息，请参阅 local.settings.file。
• requirements.txt：包含在发布到 Azure 时系统安装的 Python 包列表。
• host.json：包含在函数应用实例中影响所有函数的配置选项。 此文件会被发布到 Azure。 本地运行时，并非所有选项都受支持。 若要了解详细信息，请参阅 host.json。
• .vscode/：（可选）包含存储的 Visual Studio Code 配置。 若要了解详细信息，请参阅 Visual Studio Code 设置。
• .venv/：（可选）包含本地开发使用的 Python 虚拟环境。
• Dockerfile：（可选）在自定义容器中发布项目时使用。
• tests/：（可选）包含函数应用的测试用例。
• .funcignore：（可选）声明不应发布到 Azure 的文件。 通常，此文件包含 .vscode/ 以忽略编辑器设置，包含 .venv/ 以忽略本地 Python 虚拟环境，包含 tests/ 以忽略测试用例，包含 local.settings.json 以阻止发布本地应用设置。

每个函数都有自己的代码文件和绑定配置文件 function.json。

Python 函数项目的建议文件夹结构如以下示例所示：

 <project_root>/
 | - .venv/
 | - .vscode/
 | - function_app.py
 | - additional_functions.py
 | - tests/
 | | - test_my_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

主项目文件夹 <project_root> 可以包含以下文件：

• .venv/：（可选）包含本地开发使用的 Python 虚拟环境。
• .vscode/：（可选）包含存储的 Visual Studio Code 配置。 若要了解详细信息，请参阅 Visual Studio Code 设置。
• function_app.py：所有函数及其相关触发器和绑定的默认位置。
• additional_functions.py：（可选）包含通过蓝图在 function_app.py 中引用的函数（通常用于逻辑分组）的任何其他 Python 文件。
• tests/：（可选）包含函数应用的测试用例。
• .funcignore：（可选）声明不应发布到 Azure 的文件。 通常，此文件包含 .vscode/ 以忽略编辑器设置，包含 .venv/ 以忽略本地 Python 虚拟环境，包含 tests/ 以忽略测试用例，包含 local.settings.json 以阻止发布本地应用设置。
• host.json：包含在函数应用实例中影响所有函数的配置选项。 此文件会被发布到 Azure。 本地运行时，并非所有选项都受支持。 若要了解详细信息，请参阅 host.json。
• local.settings.json：用于在本地运行时存储应用设置和连接字符串。 此文件不会被发布到 Azure。 若要了解详细信息，请参阅 local.settings.file。
• requirements.txt：包含在发布到 Azure 时系统安装的 Python 包列表。
• Dockerfile：（可选）在自定义容器中发布项目时使用。

蓝图

Python v2 编程模型引入了蓝图的概念。 蓝图是一个实例化的新类，用于在核心函数应用程序之外注册函数。 蓝图实例中注册的函数不直接由函数运行时编制索引。 若要为这些蓝图函数编制索引，函数应用需要从蓝图实例注册这些函数。

使用蓝图可获得以下好处：

• 可以将函数应用分解为模块化组件，从而可以在多个 Python 文件中定义函数，并将其划分为每个文件的不同组件。
• 提供可扩展的公共函数应用接口用于生成和重用你自己的 API。

以下示例演示了如何使用蓝图：

首先，在 http_blueprint.py 文件中定义 HTTP 触发的函数并将其添加到蓝图对象。

import logging 

import azure.functions as func 

bp = func.Blueprint() 

@bp.route(route="default_template") 
def default_template(req: func.HttpRequest) -> func.HttpResponse: 
    logging.info('Python HTTP trigger function processed a request.') 

    name = req.params.get('name') 
    if not name: 
        try: 
            req_body = req.get_json() 
        except ValueError: 
            pass 
        else: 
            name = req_body.get('name') 

    if name: 
        return func.HttpResponse( 
            f"Hello, {name}. This HTTP-triggered function " 
            f"executed successfully.") 
    else: 
        return func.HttpResponse( 
            "This HTTP-triggered function executed successfully. " 
            "Pass a name in the query string or in the request body for a" 
            " personalized response.", 
            status_code=200 
        ) 

接下来，在 function_app.py 文件中，导入蓝图对象并将其函数注册到函数应用。

import azure.functions as func 
from http_blueprint import bp

app = func.FunctionApp() 

app.register_functions(bp) 

导入行为

可以使用绝对引用和相对引用在函数代码中导入模块。 根据上面所述的文件夹结构，以下导入在函数文件 <project_root>\my_first_function\__init__.py 中工作：

from shared_code import my_first_helper_function #(absolute)

import shared_code.my_second_helper_function #(absolute)

from . import example #(relative)

以下 __app__ 导入和 beyond top-level 相对导入已弃用，因为它们不受静态类型检查器支持，且不受 Python 测试框架支持：

from __app__.shared_code import my_first_helper_function #(deprecated __app__ import)

from ..shared_code import my_first_helper_function #(deprecated beyond top-level relative import)

在 Azure Functions 中，输入分为两种类别：触发器输入和其他输入。 虽然它们在 function.json 文件中并不相同，但它们在 Python 代码中的使用方法却是相同的。 在本地运行时，触发器和输入源的连接字符串或机密映射到 local.settings.json 文件中的值；在 Azure 中运行时，它们映射到应用程序设置。

例如，以下代码演示了两个输入之间的差异：

// function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous",
      "route": "items/{id}"
    },
    {
      "name": "obj",
      "direction": "in",
      "type": "blob",
      "path": "samples/{id}",
      "connection": "STORAGE_CONNECTION_STRING"
    }
  ]
}

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>"
  }
}

# __init__.py
import azure.functions as func
import logging

def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

调用函数时，HTTP 请求作为 req 传递给函数。 将基于路由 URL 中的 ID 从 Azure Blob 存储帐户检索一个条目，并在函数体中将其用作 obj。 在这里，指定的存储帐户是在 CONNECTION_STRING 应用设置中找到的连接字符串。

在 Azure Functions 中，输入分为两种类别：触发器输入和其他输入。 尽管它们是使用不同的修饰器定义的，但在 Python 代码中的用法相似。 在本地运行时，触发器和输入源的连接字符串或机密映射到 local.settings.json 文件中的值；在 Azure 中运行时，它们映射到应用程序设置。

例如，以下代码演示了如何定义 Blob 存储输入绑定：

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>",
    "AzureWebJobsFeatureFlags": "EnableWorkerIndexing"
  }
}

# function_app.py
import azure.functions as func
import logging

app = func.FunctionApp()

@app.route(route="req")
@app.read_blob(arg_name="obj", path="samples/{id}", 
               connection="STORAGE_CONNECTION_STRING")
def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

调用函数时，HTTP 请求作为 req 传递给函数。 将基于路由 URL 中的 ID 从 Azure Blob 存储帐户检索一个条目，并在函数体中将其用作 obj。 在这里，指定的存储帐户是在 STORAGE_CONNECTION_STRING 应用设置中找到的连接字符串。

输出可以在返回值和输出参数中进行表示。 如果只有一个输出，则建议使用返回值。 对于多个输出，必须使用输出参数。

若要使用函数的返回值作为输出绑定的值，则绑定的 name 属性应在 function.json 文件中设置为 $return。

若要生成多个输出，请使用 azure.functions.Out 接口提供的 set() 方法将值分配给绑定。 例如，以下函数可以将消息推送到队列，还可返回 HTTP 响应。

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous"
    },
    {
      "name": "msg",
      "direction": "out",
      "type": "queue",
      "queueName": "outqueue",
      "connection": "STORAGE_CONNECTION_STRING"
    },
    {
      "name": "$return",
      "direction": "out",
      "type": "http"
    }
  ]
}

import azure.functions as func

def main(req: func.HttpRequest,
         msg: func.Out[func.QueueMessage]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

输出可以在返回值和输出参数中进行表示。 如果只有一个输出，则建议使用返回值。 对于多个输出，必须使用输出参数。

若要生成多个输出，请使用 azure.functions.Out 接口提供的 set() 方法将值分配给绑定。 例如，以下函数可以将消息推送到队列，还可返回 HTTP 响应。

# function_app.py
import azure.functions as func

app = func.FunctionApp()

@app.write_blob(arg_name="msg", path="output-container/{name}",
                connection="CONNECTION_STRING")
def test_function(req: func.HttpRequest,
                  msg: func.Out[str]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

HTTP 触发器在 function.json 文件中定义。 绑定的 name 必须与函数中的命名参数匹配。 前面的示例中使用了绑定名称 req。 此参数是 HttpRequest 对象，并返回 HttpResponse 对象。

从 HttpRequest 对象中，可以获取请求标头、查询参数、路由参数和消息正文。

以下示例来自适用于 Python 的 HTTP 触发器模版。

def main(req: func.HttpRequest) -> func.HttpResponse:
    headers = {"my-http-header": "some-value"}

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello {name}!", headers=headers)
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             headers=headers, status_code=400
        )

在此函数中，从 HttpRequest 对象的 params 参数获取 name 查询参数的值。 使用 get_json 方法读取 JSON 编码的消息正文。

同样，你可以在返回的 HttpResponse 对象中为响应消息设置 status_code 和 headers。

HTTP 触发器定义为一种方法，该方法采用命名绑定参数（即 HttpRequest 对象）并返回 HttpResponse 对象。 请将 function_name 装饰器应用于方法来定义函数名称，同时通过应用 route 装饰器来设置 HTTP 终结点。

以下示例取自 Python v2 编程模型的 HTTP 触发器模板，其中的绑定参数名称为 req。 它是使用 Azure Functions Core Tools 或 Visual Studio Code 创建函数时提供的示例代码。

@app.function_name(name="HttpTrigger1")
@app.route(route="hello")
def test_function(req: func.HttpRequest) -> func.HttpResponse:
     logging.info('Python HTTP trigger function processed a request.')

     name = req.params.get('name')
     if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

     if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP-triggered function executed successfully.")
     else:
        return func.HttpResponse(
             "This HTTP-triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

从 HttpRequest 对象中，可以获取请求标头、查询参数、路由参数和消息正文。 在此函数中，从 HttpRequest 对象的 params 参数获取 name 查询参数的值。 使用 get_json 方法读取 JSON 编码的消息正文。

同样，你可以在返回的 HttpResponse 对象中为响应消息设置 status_code 和 headers。

若要在此示例中传入名称，请粘贴运行函数时提供的 URL，并在其后面追加 "?name={name}"。

可以将 Web 服务器网关接口 (WSGI) 兼容的框架和异步服务器网关接口 (ASGI) 兼容的框架（例如 Flask 和 FastAPI）用于 HTTP 触发的 Python 函数。 本部分介绍如何修改函数以支持这些框架。

首先，必须更新 function.json 文件，使其在 HTTP 触发器中包含 route，如以下示例中所示：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
       "authLevel": "anonymous",
       "type": "httpTrigger",
       "direction": "in",
       "name": "req",
       "methods": [
           "get",
           "post"
       ],
       "route": "{*route}"
    },
    {
       "type": "http",
       "direction": "out",
       "name": "$return"
    }
  ]
}

还必须更新 host.json 文件，使其包含 HTTP routePrefix，如以下示例中所示：

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

根据框架使用的接口，更新 Python 代码文件 init.py。 下面的示例展示了 ASGI 处理程序方法或 Flask 的 WSGI 包装器方法：

app = fastapi.FastAPI()

@app.get("hello/{name}")
async def get_name(name: str):
  return {"name": name}

def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return func.AsgiMiddleware(app).handle(req, context)

app = Flask("Test")

@app.route("hello/<name>", methods=["GET"])
def hello(name: str):
    return f"hello {name}"

def main(req: func.HttpRequest, context) -> func.HttpResponse:
  logging.info("Python HTTP trigger function processed a request.")
  return func.WsgiMiddleware(app).handle(req, context)

可以将异步服务器网关接口 (ASGI) 兼容的框架和 Web 服务器网关接口 (WSGI) 兼容的框架（例如 Flask 和 FastAPI）用于 HTTP 触发的 Python 函数。 必须先更新 host.json 文件以包含 HTTP routePrefix，如以下示例所示：

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[2.*, 3.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

框架代码如以下示例所示：

# function_app.py

import azure.functions as func 
from fastapi import FastAPI, Request, Response 

fast_app = FastAPI() 

@fast_app.get("/return_http_no_body") 
async def return_http_no_body(): 
    return Response(content="", media_type="text/plain") 

app = func.AsgiFunctionApp(app=fast_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

# function_app.py

import azure.functions as func 
from flask import Flask, Response 

flask_app = Flask(__name__) 

@flask_app.get("/return_http") 
def return_http(): 
    return Response("<h1>Hello World™</h1>", mimetype="text/html") 

app = func.WsgiFunctionApp(app=flask_app.wsgi_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

在 Azure Functions 中，服务连接字符串等应用程序设置在运行时将公开为环境变量。 在代码中访问这些设置有两种主要方法。

这两种方法都需要声明 import os。

以下示例使用名为 myAppSetting 的键通过 os.environ["myAppSetting"] 获取应用程序设置：

import logging
import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

对于本地开发，应用程序设置在 local.settings.json 文件中维护。

在 Azure Functions 中，服务连接字符串等应用程序设置在运行时将公开为环境变量。 在代码中访问这些设置有两种主要方法。

这两种方法都需要声明 import os。

以下示例使用名为 myAppSetting 的键通过 os.environ["myAppSetting"] 获取应用程序设置：

import logging
import os

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

对于本地开发，应用程序设置在 local.settings.json 文件中维护。

使用新的编程模型时，请在 local.settings.json 文件中启用以下应用设置，如下所示：

"AzureWebJobsFeatureFlags": "EnableWorkerIndexing"

部署函数时不会自动创建此设置。 必须在 Azure 上的函数应用中显式创建此设置，才能使用 v2 模型运行此设置。

首先创建 <project_root>/my_second_function/function.json 文件并将此函数定义为 HTTP 触发器。

{
  "scriptFile": "__init__.py",
  "entryPoint": "main",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

接下来，可以实现 my_second_function 和 shared_code.my_second_helper_function。

# <project_root>/my_second_function/__init__.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

# Define an HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
def main(req: func.HttpRequest) -> func.HttpResponse:
  logging.info('Executing my_second_function.')

  initial_value: int = int(req.params.get('value'))
  doubled_value: int = my_second_helper_function.double(initial_value)

  return func.HttpResponse(
    body=f"{initial_value} * 2 = {doubled_value}",
    status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

可以开始为 HTTP 触发器编写测试用例。

# <project_root>/tests/test_my_second_function.py
import unittest

import azure.functions as func
from my_second_function import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    resp = main(req)

    # Check the output.
    self.assertEqual(resp.get_body(), b'21 * 2 = 42',)

在 .venv Python 虚拟环境文件夹中安装你偏好的 Python 测试框架，例如 pip install pytest。 然后运行 pytest tests 即可检查测试结果。

首先创建 <project_root>/function_app.py 文件，并将 my_second_function 函数实现为 HTTP 触发器和 shared_code.my_second_helper_function。

# <project_root>/function_app.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

app = func.FunctionApp()

# Define the HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
@app.function_name(name="my_second_function")
@app.route(route="hello")
def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Executing my_second_function.')

    initial_value: int = int(req.params.get('value'))
    doubled_value: int = my_second_helper_function.double(initial_value)

    return func.HttpResponse(
        body=f"{initial_value} * 2 = {doubled_value}",
        status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

可以开始为 HTTP 触发器编写测试用例。

# <project_root>/tests/test_my_second_function.py
import unittest
import azure.functions as func

from function_app import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    func_call = main.build().get_user_function()
    resp = func_call(req)
    # Check the output.
    self.assertEqual(
        resp.get_body(),
        b'21 * 2 = 42',
    )

在 .venv Python 虚拟环境文件夹中安装你偏好的 Python 测试框架，例如 pip install pytest。 然后运行 pytest tests 即可检查测试结果。