Vývoj rozšíření pracovních procesů Pythonu pro Azure Functions

Poznámka:

Počínaje Pythonem 3.13 se rozšíření pracovních procesů Pythonu už nebudou podporovat.

Azure Functions umožňuje integrovat vlastní chování v rámci spouštění funkcí Pythonu. Tato funkce umožňuje vytvářet obchodní logiku, kterou zákazníci můžou snadno používat ve svých vlastních aplikacích funkcí. Rozšíření pracovních procesů jsou podporována v programovacích modelech v1 i v2 Pythonu.

V tomto kurzu se naučíte:

Vytvořte rozšíření pracovního procesu Pythonu na úrovni aplikace pro Azure Functions.
Používejte své rozšíření v aplikaci tak, jak to používají vaši zákazníci.
Zabalte a publikujte rozšíření k použití.

Požadavky

Než začnete, musíte splnit tyto požadavky:

Python 3.7 nebo vyšší Úplný seznam podporovaných verzí Pythonu ve službě Azure Functions najdete v příručce pro vývojáře Pythonu.
Nástroje Azure Functions Core Tools verze 4.0.5095 nebo novější, které podporují použití rozšíření s programovacím modelem Pythonu v2. Zkontrolujte svou verzi pomocí func --version.
Visual Studio Code nainstalovaný na jedné z podporovaných platforem.

Vytvoření rozšíření Python Worker

Rozšíření, které vytváříte, hlásí uplynulý čas vyvolání triggeru HTTP v protokolech konzoly a v textu odpovědi HTTP.

Struktura složek

Složka projektu rozšíření by měla vypadat jako následující struktura:

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

Složka nebo soubor	Description
.venv/	(Volitelné) Obsahuje virtuální prostředí Pythonu používané pro místní vývoj.
python_worker_extension/	Obsahuje zdrojový kód rozšíření pracovního procesu Pythonu. Tato složka obsahuje hlavní modul Pythonu, který se má publikovat do PyPI.
setup.py	Obsahuje metadata balíčku rozšíření pracovního procesu Pythonu.
readme.md	Obsahuje instrukce a použití vašeho rozšíření. Tento obsah se zobrazí jako popis na domovské stránce projektu PyPI.

Konfigurace metadat projektu

Nejprve vytvoříte setup.py, který poskytuje základní informace o vašem balíčku. Abyste měli jistotu, že je vaše rozšíření správně distribuováno a integrováno do funkčních aplikací zákazníka, ověřte, že 'azure-functions >= 1.7.0, < 2.0.0' je v install_requires části.

V následující šabloně byste měli podle potřeby změnit authorpole , author_email, install_requireslicense, , packagesa url pole.

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,
)

Dále implementujete váš kód rozšíření na úrovni aplikace.

Implementace rozšíření časovače

Přidejte následující kód python_worker_extension_timer/__init__.py pro implementaci rozšíření na úrovni aplikace:

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()

Tento kód dědí z AppExtensionBase tak, aby rozšíření platilo pro každou funkci v aplikaci. Rozšíření můžete také implementovat v oboru na úrovni funkce tak, že zdědíte z FuncExtensionBase.

Metoda init je metoda třídy, která je volána pracovním procesem při importu třídy rozšíření. Tady můžete pro rozšíření provádět akce inicializace. V tomto případě se inicializuje mapa hash pro záznam počátečního času vyvolání pro každou funkci.

Metoda configure je určená pro zákazníky. V souboru readme můžete zákazníkům sdělit, kdy potřebují volat Extension.configure(). Soubor readme by měl také zdokumentovat možnosti rozšíření, možnou konfiguraci a využití vašeho rozšíření. V tomto příkladu si zákazníci mohou zvolit, jestli je uplynulý čas hlášen v HttpResponse.

Metoda pre_invocation_app_level je volána pracovním procesem Pythonu před spuštěním funkce. Poskytuje informace z funkce, jako je kontext funkce a argumenty. V tomto příkladu rozšíření zaznamená zprávu a zaznamená počáteční čas vyvolání na základě invocation_id.

Podobně se post_invocation_app_level volá po provedení funkce. Tento příklad vypočítá uplynulý čas na základě času zahájení a aktuálního času. Také přepíše návratovou hodnotu odpovědi HTTP.

Vytvořte readme.md

V kořenovém adresáři projektu rozšíření vytvořte soubor readme.md. Tento soubor obsahuje pokyny a použití vašeho rozšíření. Obsah readme.md se zobrazí jako popis na domovské stránce projektu 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.

Použijte rozšíření lokálně

Teď, když jste vytvořili rozšíření, můžete ho použít v projektu aplikace a ověřit, jestli funguje podle očekávání.

Vytvoření funkce triggeru HTTP

Vytvořte novou složku pro projekt aplikace a přejděte na ni.
Z příslušného prostředí, jako je Bash, spusťte následující příkaz pro inicializaci projektu:
```
func init --python
```
Pomocí následujícího příkazu vytvořte novou funkci triggeru HTTP, která umožňuje anonymní přístup:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Aktivace virtuálního prostředí

Vytvořte virtuální prostředí Pythonu na základě operačního systému následujícím způsobem:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Aktivujte virtuální prostředí Pythonu na základě operačního systému následujícím způsobem:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Konfigurace rozšíření

Pomocí následujícího příkazu nainstalujte vzdálené balíčky pro projekt aplikace funkcí:
```
pip install -r requirements.txt
```
Nainstalujte rozšíření z místního souborového umístění v upravitelném režimu takto:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
V tomto příkladu nahraďte <PYTHON_WORKER_EXTENSION_ROOT> umístěním kořenového souboru vašeho projektu rozšíření.

Když zákazník použije vaše rozšíření, místo toho přidá umístění vašeho balíčku rozšíření do souboru requirements.txt, jak je znázorněno v následujících příkladech:
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
Otevřete soubor projektu local.settings.json a přidejte následující pole do Values:
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Při spuštění v Azure místo toho přidáte PYTHON_ENABLE_WORKER_EXTENSIONS=1 do nastavení aplikace v aplikaci funkcí.
Před funkci v souboru main pro programovací model v1 nebo do souboru function_app.py programovacího modelu v2 přidejte následující dva řádky:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Tento kód naimportuje TimerExtension modul a nastaví append_to_http_response konfigurační hodnotu.

Ověřte rozšíření

V kořenové složce projektu aplikace spusťte hostitele funkce pomocí func host start --verbose. Ve výstupu byste měli vidět místní koncový bod funkce jako https://localhost:7071/api/HttpTrigger.
V prohlížeči odešlete požadavek GET na https://localhost:7071/api/HttpTriggeradresu . Měla by se zobrazit odpověď podobná následující s připojenými daty TimeElapsed pro požadavek.
```
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)
```

Publikování rozšíření

Po vytvoření a ověření rozšíření stále potřebujete dokončit tyto zbývající úlohy publikování:

Zvolte licenci.
Vytvořte readme.md a další dokumentaci.
Publikujte knihovnu rozšíření do registru balíčků Pythonu nebo do systému správy verzí (VCS).

PyPI
GitHub

Publikování rozšíření do PyPI:

Spuštěním následujícího příkazu nainstalujte twine a wheel do výchozího prostředí Pythonu nebo do virtuálního prostředí:
```
pip install twine wheel
```
Odeberte starou dist/ složku z úložiště rozšíření.
Spuštěním následujícího příkazu vygenerujte nový balíček uvnitř dist/:
```
python setup.py sdist bdist_wheel
```
Spuštěním následujícího příkazu nahrajte balíček do PyPI:
```
twine upload dist/*
```
Při nahrávání možná budete muset zadat přihlašovací údaje k účtu PyPI. Nahrání balíčku můžete také otestovat pomocí twine upload -r testpypi dist/*. Další informace najdete v dokumentaci k Twine.

Po provedení těchto kroků můžou zákazníci vaše rozšíření použít tak, že do svého requirements.txtzadají název vašeho balíčku.

Další informace najdete v oficiálním kurzu balíčkování Pythonu.

Examples

Dokončený ukázkový projekt rozšíření si můžete prohlédnout z tohoto článku v python_worker_extension_timer ukázkovém úložišti.
Integrace OpenCensus je opensourcový projekt, který pomocí rozšiřujícího rozhraní integruje trasování telemetrie v aplikacích Azure Functions v Pythonu. Informace o implementaci tohoto rozšíření pracovního procesu Pythonu najdete v úložišti opencensus-python-extensions-azure .

Další kroky

Další informace o vývoji v Pythonu pro Azure Functions najdete v následujících zdrojích informací:

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-01-16