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 应用设置中找到的连接字符串。

Functions 支持用于 Azure Blob 存储的 Python SDK 类型绑定，这使你可以使用基础 BlobClient 类型处理 blob 数据。

先决条件

• Azure Functions 运行时版本 4.34 或更高版本。
• Python 版本 3.9 或支持的更高版本。

为 Blob 存储扩展启用 SDK 类型绑定

1. 将 azurefunctions-extensions-bindings-blob 扩展包添加到项目的 requirements.txt 文件中，其中至少应包括以下包：

   azure-functions
   azurefunctions-extensions-bindings-blob
   

2. 将以下代码添加到项目的 function_app.py 文件中，将导入 SDK 类型绑定：

   import azurefunctions.extensions.bindings.blob as blob
   

SDK 类型绑定示例

以下示例演示如何从 Blob 存储触发器 (blob_trigger) 和 HTTP 触发器 (blob_input) 上的输入绑定获取 BlobClient：

import logging

import azure.functions as func
import azurefunctions.extensions.bindings.blob as blob

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.blob_trigger(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_trigger(client: blob.BlobClient):
    logging.info(
        f"Python blob trigger function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )


@app.route(route="file")
@app.blob_input(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_input(req: func.HttpRequest, client: blob.BlobClient):
    logging.info(
        f"Python blob input function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )
    return "ok"

可以在 Python 扩展存储库中查看 Blob 存储的其他 SDK 类型绑定示例：

• ContainerClient 类型
• StorageStreamDownloader 类型

先决条件

• Azure Functions 运行时版本 4.34.1 或更高版本。
• Python 版本 3.8 或支持的更高版本。

启用 HTTP 流

默认已禁用 HTTP 流。 需要在应用程序设置中启用此功能，并更新代码以使用 FastAPI 包。

1. 将 azurefunctions-extensions-http-fastapi 扩展包添加到项目的 requirements.txt 文件中，其中至少应包括以下包：

   azure-functions
   azurefunctions-extensions-http-fastapi
   

2. 将以下代码添加到项目的 function_app.py 文件中，将导入 FastAPI 扩展：

   from azurefunctions.extensions.http.fastapi import Request, StreamingResponse
   

3. 部署到 Azure 时，请在函数应用中添加以下应用程序设置：

   • "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"
   • "PYTHON_ENABLE_INIT_INDEXING": "1"

   在本地运行时，还需要将同样的设置添加到 local.settings.json 项目文件。

HTTP 流示例

启用 HTTP 流式处理功能后，可以创建通过 HTTP 流式传输数据的函数。

以下示例是一个 HTTP 触发的函数，用于流式传输 HTTP 响应数据。 可以使用这些功能来支持诸如通过管道发送事件数据以实时可视化或检测大量数据中的异常并提供即时通知等场景。

import time

import azure.functions as func
from azurefunctions.extensions.http.fastapi import Request, StreamingResponse

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


def generate_sensor_data():
    """Generate real-time sensor data."""
    for i in range(10):
        # Simulate temperature and humidity readings
        temperature = 20 + i
        humidity = 50 + i
        yield f"data: {{'temperature': {temperature}, 'humidity': {humidity}}}\n\n"
        time.sleep(1)


@app.route(route="stream", methods=[func.HttpMethod.GET])
async def stream_sensor_data(req: Request) -> StreamingResponse:
    """Endpoint to stream real-time sensor data."""
    return StreamingResponse(generate_sensor_data(), media_type="text/event-stream")

以下示例是一个 HTTP 触发的函数，用于实时接收和处理来自客户端的流数据。 它演示了流式上传功能，此功能对于处理连续数据流和处理 IoT 设备中的事件数据等场景非常有用。

import azure.functions as func
from azurefunctions.extensions.http.fastapi import JSONResponse, Request

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.route(route="streaming_upload", methods=[func.HttpMethod.POST])
async def streaming_upload(req: Request) -> JSONResponse:
    """Handle streaming upload requests."""
    # Process each chunk of data as it arrives
    async for chunk in req.stream():
        process_data_chunk(chunk)

    # Once all data is received, return a JSON response indicating successful processing
    return JSONResponse({"status": "Data uploaded and processed successfully"})


def process_data_chunk(chunk: bytes):
    """Process each data chunk."""
    # Add custom processing logic here
    pass

调用 HTTP 流

必须使用 HTTP 客户端库对函数的 FastAPI 终结点进行流式处理调用。 你使用的客户端工具或浏览器可能无法本机支持流式处理，或者只能返回第一个数据块。

可以使用如下所示的客户端脚本将流式处理数据发送到 HTTP 终结点：

import httpx # Be sure to add 'httpx' to 'requirements.txt'
import asyncio

async def stream_generator(file_path):
    chunk_size = 2 * 1024  # Define your own chunk size
    with open(file_path, 'rb') as file:
        while chunk := file.read(chunk_size):
            yield chunk
            print(f"Sent chunk: {len(chunk)} bytes")

async def stream_to_server(url, file_path):
    timeout = httpx.Timeout(60.0, connect=60.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(url, content=stream_generator(file_path))
        return response

async def stream_response(response):
    if response.status_code == 200:
        async for chunk in response.aiter_raw():
            print(f"Received chunk: {len(chunk)} bytes")
    else:
        print(f"Error: {response}")

async def main():
    print('helloworld')
    # Customize your streaming endpoint served from core tool in variable 'url' if different.
    url = 'http://localhost:7071/api/streaming_upload'
    file_path = r'<file path>'

    response = await stream_to_server(url, file_path)
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

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

若要使用函数的返回值作为输出绑定的值，则绑定的 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 即可检查测试结果。