你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

为 Azure Functions 开发 Python 辅助角色扩展

注释

从 Python 3.13 开始，将不再支持 Python 辅助角色扩展。

Azure Functions 允许将自定义行为作为 Python 函数执行的一部分进行集成。借助此功能，可以创建客户可在其自己的函数应用中轻松使用的业务逻辑。 v1 和 v2 Python 编程模型都支持辅助角色扩展。

在本教程中，你将学习如何：

为 Azure Functions 创建应用程序级 Python 辅助角色扩展。
在应用中像客户那样使用该扩展。
打包并发布扩展以供使用。

先决条件

在开始之前，必须满足以下要求：

Python 3.7 或更高版本。要查看 Azure Functions 支持的 Python 版本的完整列表，请参阅 Python 开发人员指南。
Azure Functions Core Tools 版本 4.0.5095 或更高版本，它支持将扩展与 v2 Python 编程模型配合使用。使用 func --version 检查版本。
已在某个支持的平台上安装 Visual Studio Code。

创建 Python 工作器扩展

创建的扩展报告控制台日志和 HTTP 响应正文中 HTTP 触发器调用的已用时间。

文件夹结构

扩展项目的文件夹应类似于以下结构：

<python_worker_extension_root>/
 | - .venv/
 | - python_worker_extension_timer/
 | | - __init__.py
 | - setup.py
 | - readme.md

文件夹/文件	Description
.venv/	（可选）包含用于本地开发的 Python 虚拟环境。
python_worker_extension/	包含 Python 工作线程扩展的源代码。此文件夹包含要发布到 PyPI 的主 Python 模块。
setup.py	包含 Python 工作程序扩展包的元数据。
readme.md	包含您扩展的说明与用法。此内容在 PyPI 项目的主页中显示为说明。

配置项目元数据

首先创建 setup.py，它提供有关软件包的基本信息。确保扩展已正确部署并集成到客户的函数应用中，请确认'azure-functions >= 1.7.0, < 2.0.0'位于install_requires部分。

在以下模板中，应根据需要更改 author、 author_email、 install_requires、 license、 packages和 url 字段。

from setuptools import find_packages, setup
setup(
    name='python-worker-extension-timer',
    version='1.0.0',
    author='Your Name Here',
    author_email='your@email.here',
    classifiers=[
        'Intended Audience :: End Users/Desktop',
        'Development Status :: 5 - Production/Stable',
        'Intended Audience :: End Users/Desktop',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: 3.9',
        'Programming Language :: Python :: 3.10',
    ],
    description='Python Worker Extension Demo',
    include_package_data=True,
    long_description=open('readme.md').read(),
    install_requires=[
        'azure-functions >= 1.7.0, < 2.0.0',
        # Any additional packages that will be used in your extension
    ],
    extras_require={},
    license='MIT',
    packages=find_packages(where='.'),
    url='https://your-github-or-pypi-link',
    zip_safe=False,
)

接下来，你将在应用程序级别范围内实现扩展代码。

实现计时器扩展

添加以下代码 python_worker_extension_timer/__init__.py 以实现应用程序级扩展：

import typing
from logging import Logger
from time import time
from azure.functions import AppExtensionBase, Context, HttpResponse
class TimerExtension(AppExtensionBase):
    """A Python worker extension to record elapsed time in a function invocation
    """

    @classmethod
    def init(cls):
        # This records the starttime of each function
        cls.start_timestamps: typing.Dict[str, float] = {}

    @classmethod
    def configure(cls, *args, append_to_http_response:bool=False, **kwargs):
        # Customer can use TimerExtension.configure(append_to_http_response=)
        # to decide whether the elapsed time should be shown in HTTP response
        cls.append_to_http_response = append_to_http_response

    @classmethod
    def pre_invocation_app_level(
        cls, logger: Logger, context: Context,
        func_args: typing.Dict[str, object],
        *args, **kwargs
    ) -> None:
        logger.info(f'Recording start time of {context.function_name}')
        cls.start_timestamps[context.invocation_id] = time()

    @classmethod
    def post_invocation_app_level(
        cls, logger: Logger, context: Context,
        func_args: typing.Dict[str, object],
        func_ret: typing.Optional[object],
        *args, **kwargs
    ) -> None:
        if context.invocation_id in cls.start_timestamps:
            # Get the start_time of the invocation
            start_time: float = cls.start_timestamps.pop(context.invocation_id)
            end_time: float = time()
            # Calculate the elapsed time
            elapsed_time = end_time - start_time
            logger.info(f'Time taken to execute {context.function_name} is {elapsed_time} sec')
            # Append the elapsed time to the end of HTTP response
            # if the append_to_http_response is set to True
            if cls.append_to_http_response and isinstance(func_ret, HttpResponse):
                func_ret._HttpResponse__body += f' (TimeElapsed: {elapsed_time} sec)'.encode()

此代码继承自 AppExtensionBase ，以便该扩展适用于应用中的每个函数。还可以从 FuncExtensionBase 继承，在函数级别范围内实现扩展。

init 方法是在导入扩展类时由工作进程调用的类方法。可以在此处为扩展执行初始化作。在这种情况下，将初始化哈希映射以记录每个函数的调用开始时间。

该方法 configure 面向客户。在自述文件中，可以告知客户何时需要调用 Extension.configure()。自述文件还应该阐述扩展的功能、可能的配置及扩展的用法。在此示例中，客户可以选择是否在 HttpResponse 中报告运行时间。

该pre_invocation_app_level方法在函数运行之前由 Python 工作进程调用。它提供函数中的信息，例如函数上下文和参数。在此示例中，扩展会记录消息，并根据消息invocation_id记录调用的开始时间。

同样，post_invocation_app_level 在函数执行后被调用。此示例根据开始时间和当前时间计算已用时间。它还会覆盖 HTTP 响应的返回值。

创建 readme.md

在扩展项目的根目录中创建 readme.md 文件。此文件包含扩展的说明和用法。 readme.md 内容将显示为 PyPi 项目主页中的说明。

# Python Worker Extension Timer

In this file, tell your customers when they need to call `Extension.configure()`.

The readme should also document the extension capabilities, possible configuration,
and usage of your extension.

在本地使用扩展

创建扩展后，可以在应用项目中使用它来验证它是否按预期工作。

创建 HTTP 触发器函数

为应用项目创建新文件夹并导航到它。
在相应的 shell（如 Bash）中运行以下命令以初始化项目：
```
func init --python
```
使用以下命令创建新的允许匿名访问的 HTTP 触发器函数：
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

激活虚拟环境

根据 OS 创建 Python 虚拟环境，如下所示：
- Linux
- Windows操作系统
```
python3 -m venv .venv
```
```
py -m venv .venv
```
根据 OS 激活 Python 虚拟环境，如下所示：
- Linux
- Windows操作系统
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

配置扩展

使用以下命令为函数应用项目安装远程包：
```
pip install -r requirements.txt
```
在可编辑模式下，从本地文件路径安装扩展，如下所示：
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
在此示例中，请用扩展项目的根文件位置替换 <PYTHON_WORKER_EXTENSION_ROOT>。

当客户使用你的扩展时，他们会改为将你的扩展包位置添加到 requirements.txt 文件，如以下示例所示：
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
打开 local.settings.json 项目文件，并将以下字段 Values添加到：
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
在 Azure 中运行时，请将 PYTHON_ENABLE_WORKER_EXTENSIONS=1 添加到函数应用的应用设置中。
在 v1 编程模型 __init.py__ 文件或 v2 编程模型的 function_app.py 文件中的 main 函数前面添加以下两行：
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
此代码导入 TimerExtension 模块并设置 append_to_http_response 配置值。

验证扩展

在应用项目根文件夹中，启动函数主机，使用 func host start --verbose。应在输出中看到函数的本地终结点。https://localhost:7071/api/HttpTrigger

在浏览器中，将 GET 请求发送到 https://localhost:7071/api/HttpTrigger。应会看到如下所示的响应，其中追加了请求的 TimeElapsed 数据。

This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response. (TimeElapsed: 0.0009996891021728516 sec)

发布扩展

创建并验证扩展后，仍需完成以下剩余的发布任务：

选择许可证。
创建 readme.md 和其他文档。
将扩展库发布到 Python 包注册表或版本控制系统（VCS）。

PyPI
GitHub

要将您的扩展发布到 PyPI，请执行以下步骤：

运行以下命令，在默认的 Python 环境或虚拟环境中安装twine和wheel：
```
pip install twine wheel
```
从扩展存储库中删除旧 dist/ 文件夹。
运行以下命令以在 dist/ 中生成新包：
```
python setup.py sdist bdist_wheel
```
运行以下命令，将包上传到 PyPI：
```
twine upload dist/*
```
上传期间可能需要提供 PyPI 帐户凭据。还可以使用 twine upload -r testpypi dist/* 测试包上传。有关详细信息，请参阅 Twine 文档。

执行这些步骤后，客户可以通过在其 requirements.txt 中包含程序包名称来使用您的扩展。

有关详细信息，请参阅官方 Python 打包教程。

例子

可以在 python_worker_extension_timer 示例存储库中查看本文中已完成的示例扩展项目。
OpenCensus 集成是一个开源项目，它使用扩展接口在 Azure Functions Python 应用中集成遥测跟踪。请参阅 opencensus-python-extensions-azure 存储库，查看此 Python 辅助角色扩展的实现。

后续步骤

有关 Azure Functions Python 开发的详细信息，请参阅以下资源：

反馈

此页面是否有帮助？

Last updated on 2026-01-19