開發適用於 Azure Functions 的 Python 背景工作角色延伸模組

發行項
04/18/2024

Azure Functions 可讓您將自訂行為整合為 Python 函式執行的一部分。此功能可讓您建立商務邏輯，讓客戶可以輕鬆地在自己的函數應用程式中使用。若要深入了解，請參閱 Python 開發人員參考。 v1 和 v2 Python 程式設計模型中都支援背景工作角色延伸模組。

在本教學課程中，您將了解如何：

為 Azure Functions 建立應用程式層級的 Python 背景工作角色延伸模組。
以客戶的方式在應用程式中取用延伸模組。
封裝並發佈延伸模組以供取用。

必要條件

在您開始之前，您必須符合以下需求：

Python 3.7 或更新版本。若要檢查 Azure Functions 中支援 Python 版本的完整清單，請參閱 Python 開發人員指南。
Azure Functions Core Tools4.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 中報告經過的時間。

在函式執行之前，Python 背景工作角色會呼叫 pre_invocation_app_level 方法。會提供函式的資訊，例如函式內容和引數。在此範例中，延伸模組會記錄訊息，並根據其 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 觸發程序函式

為您的應用程式專案建立新的資料夾，並瀏覽到該資料夾。
從適當的殼層 (例如 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__ 檔案中的 main 函式新增下列兩行，或針對 v2 程式設計模型，在 function_app.py 檔案中新增下列兩行：
```
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 開發的詳細資訊，請參閱下列資源：