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

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:

Wymagania wstępne

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

• Język Python w wersji 3.7 lub nowszej. Aby sprawdzić pełną listę obsługiwanych wersji języka Python w 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 języku Python w wersji 2. Sprawdź wersję przy użyciu polecenia func --version.

• Visual Studio Code zainstalowane na jednej z obsługiwanych platform.

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

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:

   • Linux
   • Windows

   python3 -m venv .venv
   

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

   • Linux
   • Windows

   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:

   • PyPI
   • GitHub

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

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:

• Przewodnik dla deweloperów usługi Azure Functions pracujących w języku Python
• Najlepsze rozwiązania dotyczące usługi Azure Functions
• Dokumentacja usługi Azure Functions dla deweloperów