Tworzenie rozszerzeń procesów roboczych języka Python dla Azure Functions

  • Artykuł

Azure Functions umożliwia integrację 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. Aby dowiedzieć się więcej, zobacz dokumentację dla deweloperów języka Python. Rozszerzenia procesów roboczych są obsługiwane zarówno w modelach programowania w wersji 1, jak i w wersji 2 języka Python.

Z tego samouczka dowiesz się, jak wykonywać następujące czynności:

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

Wymagania wstępne

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

Tworzenie rozszerzenia procesu roboczego języka Python

Rozszerzenie, które tworzy, raportuje upłynął czas wywołania wyzwalacza HTTP w dziennikach konsoli i w treści odpowiedzi HTTP.

Struktura folderów

Folder projektu rozszerzenia powinien przypominać następującą strukturę:

<python_worker_extension_root>/
 | - .venv/
 | - python_worker_extension_timer/
 | | - __init__.py
 | - setup.py
 | - readme.md
Folder/plik Opis
.venv/ (Opcjonalnie) Zawiera środowisko wirtualne języka Python używane do programowania lokalnego.
python_worker_extension/ Zawiera kod źródłowy rozszerzenia procesu 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 instrukcje i użycie rozszerzenia. Ta zawartość jest wyświetlana jako opis na stronie głównej w projekcie PyPI.

Konfigurowanie metadanych projektu

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

W poniższym szablonie należy w razie potrzeby zmienić authorpola , author_email, install_requireslicense, 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 zakresie na poziomie 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 , aby rozszerzenie dotyczyło każdej funkcji w aplikacji. Można również zaimplementować rozszerzenie w zakresie na poziomie funkcji przez dziedziczenie z bazy danych FuncExtensionBase.

Metoda init jest metodą klasy wywoływaną przez proces roboczy podczas importowania klasy rozszerzenia. W tym miejscu możesz wykonać akcje inicjowania dla rozszerzenia. W takim przypadku mapa skrótu jest inicjowana do rejestrowania czasu rozpoczęcia wywołania dla każdej funkcji.

Metoda jest skierowana configure do klienta. W pliku readme możesz poinformować klientów o konieczności wywołania metody Extension.configure(). Plik readme powinien również udokumentować możliwości rozszerzenia, możliwe konfiguracje i użycie rozszerzenia. W tym przykładzie klienci mogą wybrać, czy upłynął czas zgłaszany w obiekcie HttpResponse.

Metoda jest wywoływana pre_invocation_app_level przez proces roboczy języka Python przed uruchomieniem funkcji. Zawiera 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 invocation_id.

Podobnie element jest wywoływany po wykonaniu post_invocation_app_level 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.

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

Korzystanie z 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

  1. Utwórz nowy folder dla projektu aplikacji i przejdź do niego.

  2. W odpowiedniej powłoce, takiej jak powłoka Bash, uruchom następujące polecenie, aby zainicjować projekt:

    func init --python

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

  1. Utwórz środowisko wirtualne języka Python na podstawie systemu operacyjnego w następujący sposób:

    python3 -m venv .venv

  2. Aktywuj środowisko wirtualne języka Python na podstawie systemu operacyjnego w następujący sposób:

    source .venv/bin/activate

Konfigurowanie rozszerzenia

  1. Zainstaluj pakiety zdalne dla projektu aplikacji funkcji przy użyciu następującego polecenia:

    pip install -r requirements.txt

  2. Zainstaluj rozszerzenie z lokalnej ścieżki pliku w trybie edytowalnym w następujący sposób:

    pip install -e <PYTHON_WORKER_EXTENSION_ROOT>

    W tym przykładzie zastąp element <PYTHON_WORKER_EXTENSION_ROOT> lokalizacją pliku głównego projektu rozszerzenia.

    Gdy klient używa rozszerzenia, zamiast tego doda lokalizację pakietu rozszerzeń do pliku requirements.txt, jak w poniższych przykładach:

    # requirements.txt
python_worker_extension_timer==1.0.0

  3. 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 zamiast tego dodasz PYTHON_ENABLE_WORKER_EXTENSIONS=1 ustawienia aplikacji w aplikacji funkcji.

  4. Dodaj następujące dwa wiersze przed main funkcją 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

  1. W folderze głównym projektu aplikacji uruchom hosta funkcji przy użyciu polecenia func host start --verbose. Powinien zostać wyświetlony lokalny punkt końcowy funkcji w danych wyjściowych jako https://localhost:7071/api/HttpTrigger.

  2. W przeglądarce wyślij żądanie GET do https://localhost:7071/api/HttpTrigger. Powinna zostać wyświetlona odpowiedź podobna do poniższej, z dołączonymi danymi TimeElapsed dla żądania.

    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ć te pozostałe zadania publikowania:

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

Aby opublikować rozszerzenie w PyPI:

  1. Uruchom następujące polecenie, aby zainstalować twine środowisko języka Python i wheel w domyślnym środowisku języka Python lub w środowisku wirtualnym:

    pip install twine wheel

  2. Usuń stary dist/ folder z repozytorium rozszerzeń.

  3. Uruchom następujące polecenie, aby wygenerować nowy pakiet wewnątrz dist/elementu :

    python setup.py sdist bdist_wheel

  4. Uruchom następujące polecenie, aby przekazać pakiet do PyPI:

    twine upload dist/*

    Podczas przekazywania 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ę twine.

Po wykonaniu tych kroków klienci mogą używać rozszerzenia, dołączając nazwę pakietu do 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 wyświetlić w tym artykule w repozytorium przykładowym python_worker_extension_timer .

  • Integracja openCensus to projekt typu open source, który używa interfejsu rozszerzenia do integracji śledzenia telemetrii w aplikacjach Azure Functions Python. Zobacz repozytorium opencensus-python-extensions-azure , aby zapoznać się z implementacją tego rozszerzenia procesu roboczego języka Python.

Następne kroki

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