Opracowywanie rozszerzeń procesów roboczych języka Python dla usługi Azure Functions

Uwaga / Notatka

Począwszy od języka Python 3.13, rozszerzenia procesu roboczego języka Python nie będą już obsługiwane.

Usługa Azure Functions umożliwia integrowanie niestandardowych zachowań w ramach wykonywania funkcji języka Python. Ta funkcja umożliwia tworzenie logiki biznesowej, której klienci mogą łatwo używać we własnych aplikacjach funkcji. Rozszerzenia procesów roboczych są obsługiwane zarówno w modelach programowania w wersji 1, jak i w wersji 2 języka Python.

W tym poradniku dowiesz się, jak:

Utwórz rozszerzenie procesu roboczego języka Python na poziomie aplikacji dla usługi Azure Functions.
Korzystaj z rozszerzenia w aplikacji tak, jak robią to Twoi klienci.
Spakuj i opublikuj rozszerzenie do użycia.

Wymagania wstępne

Przed rozpoczęciem należy spełnić następujące wymagania:

Środowisko Python w wersji 3.7 lub nowszej. Aby sprawdzić pełną listę obsługiwanych wersji języka Python w usłudze Azure Functions, zobacz Przewodnik dla deweloperów języka Python.
Narzędzia Azure Functions Core Tools w wersji 4.0.5095 lub nowszej, które obsługują używanie rozszerzenia z modelem programowania w wersji 2 języka Python. Sprawdź swoją wersję z pomocą func --version.
Program Visual Studio Code zainstalowany na jednej z obsługiwanych platform.

Utwórz rozszerzenie Worker dla Pythona

Rozszerzenie, które tworzysz, zgłasza czas upłynięcia wywołania wyzwalacza HTTP w logach konsoli oraz w treści odpowiedzi HTTP.

Struktura folderów

Folder projektu rozszerzenia powinien być podobny do następującej struktury:

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

Folder/plik	Description
.venv/	(Opcjonalnie) Zawiera środowisko wirtualne języka Python używane do programowania lokalnego.
python_worker_extension/	Zawiera kod źródłowy rozszerzenia roboczego języka Python. Ten folder zawiera główny moduł języka Python do opublikowania w interfejsie PyPI.
setup.py	Zawiera metadane pakietu rozszerzenia procesu roboczego języka Python.
readme.md	Zawiera instrukcję i użycie rozszerzenia. Ta zawartość jest wyświetlana jako opis na stronie głównej w projekcie PyPI.

Konfigurowanie metadanych projektu

Najpierw należy utworzyć element setup.py, który zawiera podstawowe informacje o twoim pakiecie. Aby upewnić się, że rozszerzenie jest dystrybuowane i poprawnie zintegrowane z aplikacjami funkcji klienta, upewnij się, że 'azure-functions >= 1.7.0, < 2.0.0' znajduje się w sekcji install_requires.

W poniższym szablonie należy w razie potrzeby zmienić pola author, author_email, install_requires, license, packages i 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,
)

Następnie zaimplementujesz kod rozszerzenia w ramach poziomu aplikacji.

Implementowanie rozszerzenia czasomierza

Dodaj następujący kod w pliku , python_worker_extension_timer/__init__.py aby zaimplementować rozszerzenie na poziomie aplikacji:

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

Ten kod dziedziczy z bazy danych AppExtensionBase , tak aby rozszerzenie dotyczyło każdej funkcji w aplikacji. Rozszerzenie można również zaimplementować w zakresie poziomu funkcji, dziedzicząc z FuncExtensionBase.

Metoda init jest metodą klasy, którą wywołuje pracownik podczas importowania klasy rozszerzenia. W tym miejscu możesz wykonać akcje inicjowania dla rozszerzenia. W takim przypadku mapowanie skrótów jest inicjowane do rejestrowania czasu rozpoczęcia wywołania dla każdej funkcji.

Metoda configure przeznaczona jest dla klientów. W pliku readme możesz poinformować klientów, kiedy muszą zadzwonić do usługi Extension.configure(). Plik readme powinien również udokumentować możliwości rozszerzenia, możliwą konfigurację i użycie rozszerzenia. W tym przykładzie klienci mogą wybrać, czy czas, który upłynął, jest zgłaszany w obiekcie HttpResponse.

Metoda pre_invocation_app_level jest wywoływana przez proces roboczy języka Python przed uruchomieniem funkcji. Udostępnia on informacje z funkcji, takie jak kontekst funkcji i argumenty. W tym przykładzie rozszerzenie rejestruje komunikat i rejestruje godzinę rozpoczęcia wywołania na podstawie jego invocation_id.

Podobnie post_invocation_app_level jest wywoływane po wykonaniu funkcji. W tym przykładzie obliczany jest czas, który upłynął na podstawie czasu rozpoczęcia i bieżącego czasu. Zastępuje również wartość zwracaną odpowiedzi HTTP.

Utwórz plik readme.md

Utwórz plik readme.md w katalogu głównym projektu rozszerzenia. Ten plik zawiera instrukcje i użycie rozszerzenia. Zawartość readme.md jest wyświetlana jako opis na stronie głównej w projekcie 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.

Używaj rozszerzenia lokalnie

Po utworzeniu rozszerzenia możesz użyć go w projekcie aplikacji, aby sprawdzić, czy działa zgodnie z oczekiwaniami.

Tworzenie funkcji wyzwalacza HTTP

Utwórz nowy folder dla projektu aplikacji i przejdź do niego.
W odpowiedniej powłoce, jak Bash, uruchom następujące polecenie, aby zainicjować projekt.
```
func init --python
```
Użyj następującego polecenia, aby utworzyć nową funkcję wyzwalacza HTTP, która umożliwia dostęp anonimowy:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Aktywowanie środowiska wirtualnego

Utwórz środowisko wirtualne języka Python na podstawie systemu operacyjnego w następujący sposób:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Aktywuj środowisko wirtualne języka Python na podstawie systemu operacyjnego w następujący sposób:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Konfigurowanie rozszerzenia

Zainstaluj pakiety zdalne dla projektu aplikacji funkcji przy użyciu następującego polecenia:
```
pip install -r requirements.txt
```
Zainstaluj rozszerzenie ze ścieżki pliku lokalnego w trybie edytowalnym w następujący sposób:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
W tym przykładzie zastąp <PYTHON_WORKER_EXTENSION_ROOT> ścieżką do pliku głównego projektu rozszerzenia.

Gdy klient używa rozszerzenia, zamiast tego doda lokalizację pakietu rozszerzenia do pliku requirements.txt, jak w następujących przykładach:
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
Otwórz plik projektu local.settings.json i dodaj następujące pole do :Values
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Podczas uruchamiania na platformie Azure należy dodać PYTHON_ENABLE_WORKER_EXTENSIONS=1 do ustawień aplikacji w aplikacji funkcji.
Dodaj następujące dwa wiersze przed funkcją main w pliku __init.py__ dla modelu programowania w wersji 1 lub w pliku function_app.py dla modelu programowania w wersji 2:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Ten kod importuje TimerExtension moduł i ustawia append_to_http_response wartość konfiguracji.

Weryfikowanie rozszerzenia

W folderze głównym projektu aplikacji uruchom hosta funkcji przy użyciu polecenia func host start --verbose. W danych wyjściowych powinien zostać wyświetlony lokalny punkt końcowy funkcji jako https://localhost:7071/api/HttpTrigger.
W przeglądarce wyślij żądanie GET do https://localhost:7071/api/HttpTrigger. Powinna zostać wyświetlona odpowiedź podobna do poniższego przykładu, z dołączonymi dla żądania danymi 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)
```

Publikowanie rozszerzenia

Po utworzeniu i zweryfikowaniu rozszerzenia nadal trzeba wykonać pozostałe zadania publikowania:

Wybierz licencję.
Utwórz plik readme.md i pozostałą dokumentację.
Opublikuj bibliotekę rozszerzeń w rejestrze pakietów języka Python lub systemie kontroli wersji (VCS).

PyPI
GitHub

Aby opublikować rozszerzenie do PyPI:

Uruchom następujące polecenie, aby zainstalować twine i wheel w domyślnym środowisku języka Python lub w środowisku wirtualnym:
```
pip install twine wheel
```
Usuń stary dist/ folder z repozytorium rozszerzeń.
Uruchom następujące polecenie, aby wygenerować nowy pakiet w programie dist/:
```
python setup.py sdist bdist_wheel
```
Uruchom następujące polecenie, aby przekazać pakiet do PyPI:
```
twine upload dist/*
```
Podczas wysyłania może być konieczne podanie poświadczeń konta PyPI. Możesz również przetestować przekazywanie pakietu za pomocą polecenia twine upload -r testpypi dist/*. Aby uzyskać więcej informacji, zobacz dokumentację usługi Twine.

Po wykonaniu tych kroków klienci mogą używać rozszerzenia, uwzględniając nazwę pakietu w requirements.txt.

Aby uzyskać więcej informacji, zobacz oficjalny samouczek dotyczący tworzenia pakietów w języku Python.

Przykłady

Ukończony przykładowy projekt rozszerzenia można zobaczyć w tym artykule w przykładowym repozytorium python_worker_extension_timer.
Integracja OpenCensus to projekt open source, który używa interfejsu rozszerzeń do integrowania śledzenia telemetrii w aplikacjach Python usługi Azure Functions. Zobacz repozytorium opencensus-python-extensions-azure , aby zapoznać się z implementacją tego rozszerzenia procesu roboczego języka Python.

Dalsze kroki

Aby uzyskać więcej informacji na temat programowania w języku Python w usłudze Azure Functions, zobacz następujące zasoby:

Sprzężenie zwrotne

Czy ta strona była pomocna?

Last updated on 2026-01-16