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

开发适用于 Azure Functions 的 Python 辅助角色扩展

项目
04/28/2023

Azure Functions 可让你将自定义行为集成为 Python 函数执行的一部分。利用此功能可以创建可供客户在其自己的函数应用中轻松使用的业务逻辑。有关详细信息，请参阅 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

文件夹/文件	说明
.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 的函数本地终结点。

在浏览器中，向 https://localhost:7071/api/HttpTrigger 发送 GET 请求。应会看到如下所示的响应，其中追加了请求的 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 开发的详细信息，请参阅以下资源：

通过