Partager via


Développer des extensions de travail Python pour Azure Functions

Note

À compter de Python 3.13, les extensions worker Python ne seront plus prises en charge.

Azure Functions vous permet d’intégrer des comportements personnalisés dans le cadre de l’exécution des fonctions Python. Cette fonctionnalité vous permet de créer une logique métier que les clients peuvent facilement utiliser dans leurs propres applications de fonction. Les extensions worker sont prises en charge dans les modèles de programmation Python v1 et v2.

Dans ce tutoriel, vous apprendrez comment le faire :

  • Créez une extension de travail Python au niveau de l’application pour Azure Functions.
  • Consommez votre extension dans une application de la façon dont vos clients le font.
  • Emballer et publier une extension pour utilisation.

Prerequisites

Avant de commencer, vous devez répondre aux exigences suivantes :

Créer l’extension Worker Python

L’extension que vous créez signale le temps écoulé d’un appel de déclencheur HTTP dans les journaux de la console et dans le corps de la réponse HTTP.

Structure de dossiers

Le dossier de votre projet d’extension doit ressembler à la structure suivante :

<python_worker_extension_root>/
 | - .venv/
 | - python_worker_extension_timer/
 | | - __init__.py
 | - setup.py
 | - readme.md
Dossier/fichier Descriptif
.venv/ (Facultatif) Contient un environnement virtuel Python utilisé pour le développement local.
python_worker_extension/ Contient le code source de l’extension Worker Python. Ce dossier contient le module Python principal à publier dans PyPI.
Setup.py Contient les métadonnées du package d’extension Worker Python.
readme.md Contient l’instruction et l’utilisation de votre extension. Ce contenu s’affiche en tant que description dans la page d’accueil de votre projet PyPI.

Configurer les métadonnées du projet

Tout d’abord, vous créez setup.py, qui fournit des informations essentielles sur votre package. Pour vous assurer que votre extension est distribuée et intégrée correctement aux applications de fonction de votre client, vérifiez qu’elle 'azure-functions >= 1.7.0, < 2.0.0' figure dans la install_requires section.

Dans le modèle suivant, vous devez modifier author, , author_emailinstall_requires, license, packageset url les champs si nécessaire.

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

Ensuite, vous allez implémenter votre code d’extension dans le contexte d'application.

Implémenter l’extension du minuteur

Ajoutez le code python_worker_extension_timer/__init__.py suivant pour implémenter l’extension au niveau de l’application :

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

Ce code hérite d’AppExtensionBase afin que l’extension s’applique à chaque fonction de l’application. Vous auriez également pu implémenter l'extension à un niveau fonctionnel en héritant de FuncExtensionBase.

La init méthode est une méthode de classe appelée par le worker lorsque la classe d’extension est importée. Vous pouvez effectuer des actions d’initialisation ici pour l’extension. Dans ce cas, une carte de hachage est initialisée pour enregistrer l’heure de début de l’appel pour chaque fonction.

La configure méthode est orientée client. Dans votre fichier lisez-moi, vous pouvez indiquer à vos clients quand ils ont besoin d’appeler Extension.configure(). Ce fichier doit également documenter les fonctionnalités de l’extension, la configuration possible et l’utilisation de votre extension. Dans cet exemple, les clients peuvent choisir si le temps écoulé est signalé dans le HttpResponse.

La pre_invocation_app_level méthode est appelée par le worker Python avant l’exécution de la fonction. Il fournit les informations de la fonction, telles que le contexte de fonction et les arguments. Dans cet exemple, l’extension enregistre un message et enregistre l’heure de début d’un appel en fonction de son invocation_id.

De même, post_invocation_app_level est appelé après l'exécution de la fonction. Cet exemple calcule le temps écoulé en fonction de l’heure de début et de l’heure actuelle. Il remplace également la valeur de retour de la réponse HTTP.

Créer un readme.md

Créez un fichier readme.md à la racine de votre projet d’extension. Ce fichier contient les instructions et l’utilisation de votre extension. Le contenu readme.md s’affiche en tant que description dans la page d’accueil de votre projet 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.

Utiliser votre extension localement

Maintenant que vous avez créé une extension, vous pouvez l’utiliser dans un projet d’application pour vérifier qu’elle fonctionne comme prévu.

Créer une fonction de déclencheur HTTP

  1. Créez un dossier pour votre projet d’application et accédez-y.

  2. À partir de l’interpréteur de commandes approprié, tel que Bash, exécutez la commande suivante pour initialiser le projet :

    func init --python
    
  3. Utilisez la commande suivante pour créer une fonction de déclencheur HTTP qui autorise l’accès anonyme :

    func new -t HttpTrigger -n HttpTrigger -a anonymous
    

Activer un environnement virtuel

  1. Créez un environnement virtuel Python, basé sur le système d’exploitation comme suit :

    python3 -m venv .venv
    
  2. Activez l’environnement virtuel Python, en fonction du système d’exploitation comme suit :

    source .venv/bin/activate
    

Configurer l’extension

  1. Installez des packages distants pour votre projet d’application de fonction à l’aide de la commande suivante :

    pip install -r requirements.txt
    
  2. Installez l’extension à partir de votre chemin d’accès de fichier local, en mode modifiable comme suit :

    pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
    

    Dans cet exemple, remplacez <PYTHON_WORKER_EXTENSION_ROOT> l’emplacement du fichier racine de votre projet d’extension.

    Lorsqu’un client utilise votre extension, il ajoute à la place l’emplacement de votre package d’extension au fichier requirements.txt, comme dans les exemples suivants :

    # requirements.txt
    python_worker_extension_timer==1.0.0
    
  3. Ouvrez le fichier projet local.settings.json et ajoutez le champ suivant à Values:

    "PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
    

    Lors de l'exécution dans Azure, vous ajoutez plutôt PYTHON_ENABLE_WORKER_EXTENSIONS=1 aux paramètres de l'application dans l'application de fonction.

  4. Ajoutez les deux lignes suivantes avant la main fonction dans le fichier __init.py__ pour le modèle de programmation v1, ou dans le fichier function_app.py du modèle de programmation v2 :

    from python_worker_extension_timer import TimerExtension
    TimerExtension.configure(append_to_http_response=True)
    

    Ce code importe le TimerExtension module et définit la valeur de append_to_http_response configuration.

Vérifier l’extension

  1. À partir du dossier racine de votre projet d'application, démarrez l'hôte de fonction à l’aide de func host start --verbose. Vous devez voir le point de terminaison local de votre fonction dans la sortie en tant que https://localhost:7071/api/HttpTrigger.

  2. Dans le navigateur, envoyez une requête GET à https://localhost:7071/api/HttpTrigger. Vous devez voir une réponse semblable à celle ci-dessous, avec les données TimeElapsed pour la requête incluses.

    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)
    

Publier votre extension

Une fois que vous avez créé et vérifié votre extension, vous devez toujours effectuer ces tâches de publication restantes :

  • Choisissez une licence.
  • Créez un fichier readme.md et d’autres documents de documentation.
  • Publiez la bibliothèque d’extensions dans un registre de packages Python ou un système de contrôle de version (VCS).

Pour publier votre extension sur PyPI :

  1. Exécutez la commande suivante pour installer twine et wheel dans votre environnement Python par défaut ou dans un environnement virtuel :

    pip install twine wheel
    
  2. Supprimez l’ancien dist/ dossier de votre référentiel d’extensions.

  3. Exécutez la commande suivante pour générer un nouveau package à l’intérieur dist/:

    python setup.py sdist bdist_wheel
    
  4. Exécutez la commande suivante pour charger le package sur PyPI :

    twine upload dist/*
    

    Vous devrez peut-être fournir vos informations d’identification de compte PyPI pendant le chargement. Vous pouvez également tester le chargement de votre package avec twine upload -r testpypi dist/*. Pour plus d’informations, consultez la documentation Twine.

Après ces étapes, les clients peuvent utiliser votre extension en incluant le nom de votre package dans leur requirements.txt.

Pour plus d’informations, consultez le didacticiel officiel sur l’empaquetage Python.

Examples

  • Vous pouvez voir le projet d'extension d'exemple terminé mentionné dans cet article dans le référentiel d'exemple python_worker_extension_timer.

  • L’intégration d’OpenCensus est un projet open source qui utilise l’interface d’extension pour intégrer le suivi des données de télémétrie dans les applications Python Azure Functions. Consultez le référentiel opencensus-python-extensions-azure pour passer en revue l’implémentation de cette extension Worker Python.

Étapes suivantes

Pour plus d’informations sur le développement Python Azure Functions, consultez les ressources suivantes :