共用方式為


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

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

在本教學課程中,您將了解如何:

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

必要條件

在您開始之前,您必須符合以下需求:

建立 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 區段中。

在下列範本中,您應該視需要變更 authorauthor_emailinstall_requireslicensepackagesurl 欄位。

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 觸發程序函式

  1. 為您的應用程式專案建立新的資料夾,並瀏覽到該資料夾。

  2. 從適當的殼層 (例如 Bash) 執行下列命令來初始化專案:

    func init --python
    
  3. 使用下列命令來建立允許匿名存取的新 HTTP 觸發程序函式:

    func new -t HttpTrigger -n HttpTrigger -a anonymous
    

啟動虛擬環境

  1. 根據 OS 建立 Python 虛擬環境,如下所示:

    python3 -m venv .venv
    
  2. 根據 OS 啟動 Python 虛擬環境,如下所示:

    source .venv/bin/activate
    

設定擴充功能

  1. 使用下列命令安裝函數應用程式專案的遠端套件:

    pip install -r requirements.txt
    
  2. 在可編輯模式中,從本機檔案路徑安裝延伸模組,如下所示:

    pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
    

    在此範例中,將 <PYTHON_WORKER_EXTENSION_ROOT> 取代為延伸模組專案的根檔案位置。

    當客戶使用您的延伸模組時,他們會改為將延伸模組套件位置新增至 requirements.txt 檔案,如下列範例所示:

    # requirements.txt
    python_worker_extension_timer==1.0.0
    
  3. 開啟 local.settings.json 專案檔,並將下欄欄位新增至 Values

    "PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
    

    在 Azure 中執行時,您會改為將 PYTHON_ENABLE_WORKER_EXTENSIONS=1 新增至函數應用程式中的應用程式設定

  4. 針對 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 組態值。

驗證延伸模組

  1. 從您的應用程式專案根資料夾,使用 func host start --verbose 啟動函式主機。 您應該會在輸出中看到函式的本機端點為 https://localhost:7071/api/HttpTrigger

  2. 在瀏覽器中,將 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:

  1. 執行下列命令,以在預設 Python 環境或虛擬環境中安裝 twinewheel

    pip install twine wheel
    
  2. 從延伸模組存放庫移除舊 dist/ 資料夾。

  3. 執行下列命令,以在 dist/ 內產生新的套件:

    python setup.py sdist bdist_wheel
    
  4. 執行下列命令以將套件上傳至 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 開發的詳細資訊,請參閱下列資源: