Entwickeln von Python-Workererweiterungen für Azure Functions

Hinweis

Ab Python 3.13 werden Python-Workererweiterungen nicht mehr unterstützt.

Mit Azure Functions können Sie benutzerdefinierte Verhaltensweisen als Teil der Ausführung der Python-Funktion integrieren. Mit diesem Feature können Sie Geschäftslogik erstellen, die Kunden ganz einfach in ihren eigenen Funktions-Apps verwenden können. In den Programmiermodellen v1 und v2 Python werden Worker-Erweiterungen unterstützt.

In diesem Tutorial lernen Sie, wie Sie:

Erstellen Sie eine Python-Workererweiterung auf Anwendungsebene für Azure Functions.
Nutzen der Erweiterung in einer App wie Ihre Kunden
Paketieren und Veröffentlichen einer Erweiterung zur Nutzung.

Voraussetzungen

Bevor Sie beginnen, müssen Sie diese Anforderungen erfüllen:

Python 3.7 oder höher. Die vollständige Liste der unterstützten Python-Versionen in Azure Functions finden Sie im Python-Entwicklerhandbuch.
Die Azure Functions Core Tools, Version 4.0.5095 oder höher, die die Verwendung der Erweiterung mit dem v2 Python-Programmiermodell unterstützt. Überprüfen Sie Ihre Version mit func --version.
Visual Studio Code auf einer der unterstützten Plattformen installiert.

Erstellen der Python Worker-Erweiterung

Die Erweiterung, die Sie erstellen, meldet die verstrichene Zeit eines HTTP-Triggeraufrufs in den Konsolenprotokollen und im HTTP-Antworttext.

Ordnerstruktur

Der Ordner für Ihr Erweiterungsprojekt sollte wie die folgende Struktur aussehen:

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

Ordner/Datei	Description
.venv/	(Optional) Enthält eine virtuelle Python-Umgebung, die für die lokale Entwicklung verwendet wird.
python_worker_extension/	Enthält den Quellcode der Python-Workererweiterung. Dieser Ordner enthält das Haupt-Python-Modul, das in PyPI veröffentlicht werden soll.
setup.py	Enthält die Metadaten des Python-Workererweiterungspakets.
readme.md	Enthält die Anweisung und Verwendung Ihrer Erweiterung. Dieser Inhalt wird als Beschreibung auf der Startseite in Ihrem PyPI-Projekt angezeigt.

Konfigurieren von Projektmetadaten

Zuerst erstellen Sie setup.py, das wichtige Informationen zu Ihrem Paket bereitstellt. Um sicherzustellen, dass Ihre Erweiterung ordnungsgemäß verteilt und in die Funktions-Apps Ihres Kunden integriert ist, vergewissern Sie sich, dass sich 'azure-functions >= 1.7.0, < 2.0.0' im install_requires-Abschnitt befindet.

In der folgenden Vorlage sollten Sie die Felder author, author_email, install_requires, license, packages und url nach Bedarf ändern.

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

Als Nächstes implementieren Sie Ihren Erweiterungscode im Anwendungsbereich.

Implementierung der Timer-Erweiterung

Fügen Sie den folgenden Code hinzu python_worker_extension_timer/__init__.py , um die Erweiterung auf Anwendungsebene zu implementieren:

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

Dieser Code erbt von AppExtensionBase , sodass die Erweiterung für jede Funktion in der App gilt. Durch Erben von FuncExtensionBase hätte die Erweiterung auch auf Funktionsebene implementiert werden können.

Die init Methode ist eine Klassenmethode, die vom Worker aufgerufen wird, wenn die Erweiterungsklasse importiert wird. Hier können Sie Initialisierungsaktionen für die Erweiterung ausführen. In diesem Fall wird eine Hashzuordnung für die Aufzeichnung der Startzeit des Aufrufs für jede Funktion initialisiert.

Die configure Methode ist kundenseitig ausgerichtet. In Ihrer Infodatei können Sie Ihren Kunden mitteilen, wann sie Extension.configure() aufrufen müssen. Die Infodatei sollte auch die Erweiterungsfunktionen, die mögliche Konfiguration und die Verwendung Ihrer Erweiterung dokumentieren. In diesem Beispiel können Kunden auswählen, ob die verstrichene Zeit angegeben wird in der HttpResponse.

Die pre_invocation_app_level Methode wird vom Python-Worker aufgerufen, bevor die Funktion ausgeführt wird. Sie stellt die Informationen aus der Funktion bereit, z. B. Funktionskontext und Argumente. In diesem Beispiel protokolliert die Erweiterung eine Nachricht und zeichnet die Startzeit eines Aufrufs basierend auf seinem invocation_id auf.

Entsprechend wird die post_invocation_app_level Funktion nach der Funktionsausführung aufgerufen. In diesem Beispiel wird die verstrichene Zeit basierend auf der Startzeit und der aktuellen Uhrzeit berechnet. Außerdem wird der Rückgabewert der HTTP-Antwort überschrieben.

Erstelle eine readme.md

Erstellen Sie eine readme.md Datei im Stammverzeichnis Ihres Erweiterungsprojekts. Diese Datei enthält die Anweisungen und die Verwendung Ihrer Erweiterung. Der readme.md Inhalt wird als Beschreibung auf der Startseite in Ihrem PyPI-Projekt angezeigt.

# 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.

Lokales Nutzen Ihrer Erweiterung

Nachdem Sie eine Erweiterung erstellt haben, können Sie sie in einem App-Projekt verwenden, um zu überprüfen, ob sie wie beabsichtigt funktioniert.

Erstellen einer HTTP-Triggerfunktion

Erstellen Sie einen neuen Ordner für Ihr App-Projekt, und navigieren Sie zu diesem.
Führen Sie in der entsprechenden Shell, z. B. Bash, den folgenden Befehl aus, um das Projekt zu initialisieren:
```
func init --python
```
Verwenden Sie den folgenden Befehl, um eine neue HTTP-Triggerfunktion zu erstellen, die anonymen Zugriff ermöglicht:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Aktivieren einer virtuellen Umgebung

Erstellen Sie eine virtuelle Python-Umgebung, die auf dem Betriebssystem wie folgt basiert:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Aktivieren Sie die virtuelle Python-Umgebung basierend auf dem Betriebssystem wie folgt:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Konfigurieren der Erweiterung

Installieren Sie Remotepakete für Ihr Funktions-App-Projekt mit dem folgenden Befehl:
```
pip install -r requirements.txt
```
Installieren Sie die Erweiterung aus Ihrem lokalen Dateipfad im bearbeitbaren Modus wie folgt:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
Ersetzen Sie <PYTHON_WORKER_EXTENSION_ROOT> in diesem Beispiel durch den Stammdateispeicherort Ihres Erweiterungsprojekts.

Wenn ein Kunde Ihre Erweiterung verwendet, wird er stattdessen den Speicherort Ihres Erweiterungspakets zur Datei requirements.txt hinzufügen, wie in den folgenden Beispielen gezeigt.
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
Öffnen Sie die local.settings.json Projektdatei, und fügen Sie das folgende Feld hinzu:Values
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Fügen Sie in Azure stattdessen PYTHON_ENABLE_WORKER_EXTENSIONS=1 zu den App-Einstellungen in der Funktions-App hinzu.
Fügen Sie die folgenden zwei Zeilen vor der Funktion in der main Datei __init.py__ für das v1-Programmiermodell oder in der datei function_app.py für das v2-Programmiermodell hinzu:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Dieser Code importiert das TimerExtension Modul und legt den append_to_http_response Konfigurationswert fest.

Überprüfen Sie die Erweiterung

Starten Sie den Funktionshost vom Stammordner des App-Projekts aus.func host start --verbose Der lokale Endpunkt Ihrer Funktion sollte in der Ausgabe als https://localhost:7071/api/HttpTriggerangezeigt werden.
Senden Sie im Browser eine GET-Anforderung an https://localhost:7071/api/HttpTrigger. Es sollte eine Antwort wie folgt angezeigt werden, wobei die TimeElapsed-Daten an die Anforderung angefügt werden.
```
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)
```

Veröffentlichen Sie Ihre Erweiterung

Nachdem Sie Ihre Erweiterung erstellt und überprüft haben, müssen Sie diese verbleibenden Veröffentlichungsaufgaben noch ausführen:

Wählen Sie eine Lizenz aus.
Erstellen Sie eine readme.md und andere Dokumentationen.
Veröffentlichen Sie die Erweiterungsbibliothek in einer Python-Paketregistrierung oder einem Versionssteuerungssystem (VCS).

PyPI
GitHub

So veröffentlichen Sie Ihre Erweiterung in PyPI:

Führen Sie den folgenden Befehl aus, um twine und wheel in Ihrer Standard-Python-Umgebung oder einer virtuellen Umgebung zu installieren:
```
pip install twine wheel
```
Entfernen Sie den alten dist/ Ordner aus Ihrem Erweiterungs-Repository.
Führen Sie den folgenden Befehl aus, um ein neues Paket in dist/ zu generieren:
```
python setup.py sdist bdist_wheel
```
Führen Sie den folgenden Befehl aus, um das Paket in PyPI hochzuladen:
```
twine upload dist/*
```
Möglicherweise müssen Sie während des Uploads Ihre PyPI-Kontoanmeldeinformationen angeben. Sie können auch Ihren Paket-Upload mit twine upload -r testpypi dist/* testen. Weitere Informationen finden Sie in der Twine-Dokumentation.

Nach diesen Schritten können Kunden Ihre Erweiterung verwenden, indem Sie ihren Paketnamen in ihre requirements.txteinschließen.

Weitere Informationen finden Sie im offiziellen Python-Paketlernprogramm.

Examples

Sie können das fertige Beispielerweiterungsprojekt aus diesem Artikel im python_worker_extension_timer Beispiel-Repository anzeigen.
Die OpenCensus-Integration ist ein Open-Source-Projekt, das die Erweiterungsschnittstelle verwendet, um die Telemetrieablaufverfolgung in Azure Functions Python-Apps zu integrieren. Sehen Sie sich das Opencensus-python-extensions-azure-Repository an, um die Implementierung dieser Python-Worker-Erweiterung zu überprüfen.

Nächste Schritte

Weitere Informationen zur Python-Entwicklung für Azure Functions finden Sie in den folgenden Ressourcen:

Feedback

War diese Seite hilfreich?

Last updated on 2026-01-16