Desenvolver extensões de trabalho python para Funções do Azure

Artigo
04/26/2023

Funções do Azure permite-lhe integrar comportamentos personalizados como parte da execução da função Python. Esta funcionalidade permite-lhe criar lógica de negócio que os clientes podem utilizar facilmente nas suas próprias aplicações de funções. Para saber mais, veja a referência para programadores do Python. As extensões de trabalho são suportadas nos modelos de programação Python v1 e v2.

Neste tutorial, irá aprender a:

Crie uma extensão de trabalho python ao nível da aplicação para Funções do Azure.
Consuma a sua extensão numa aplicação da mesma forma que os seus clientes.
Empacote e publique uma extensão para consumo.

Pré-requisitos

Antes de começar, tem de cumprir estes requisitos:

Python 3.7 ou superior. Para verificar a lista completa de versões do Python suportadas no Funções do Azure, consulte o guia para programadores do Python.
O Funções do Azure Core Tools, versão 4.0.5095 ou posterior, que suporta a utilização da extensão com o modelo de programação python v2. Verifique a sua versão com func --version.
Visual Studio Code instalado numa das plataformas suportadas.

Criar a extensão de Trabalho do Python

A extensão que criar comunica o tempo decorrido de uma invocação de acionador HTTP nos registos da consola e no corpo de resposta HTTP.

Estrutura de pastas

A pasta para o projeto de extensão deve ser semelhante à seguinte estrutura:

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

Pasta/ficheiro	Description
.venv/	(Opcional) Contém um ambiente virtual python utilizado para o desenvolvimento local.
python_worker_extension/	Contém o código fonte da extensão de trabalho do Python. Esta pasta contém o módulo python principal a publicar no PyPI.
setup.py	Contém os metadados do pacote de extensão de trabalho do Python.
readme.md	Contém a instrução e a utilização da sua extensão. Este conteúdo é apresentado como a descrição na home page do projeto PyPI.

Configurar metadados do projeto

Primeiro, crie setup.pyo , que fornece informações essenciais sobre o seu pacote. Para se certificar de que a extensão está distribuída e integrada corretamente nas aplicações de funções do cliente, confirme que 'azure-functions >= 1.7.0, < 2.0.0' está na install_requires secção.

No modelo seguinte, deve alterar authoros campos , author_email, install_requires, license, packagese, url conforme necessário.

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

Em seguida, irá implementar o código de extensão no âmbito ao nível da aplicação.

Implementar a extensão de temporizador

Adicione o seguinte código python_worker_extension_timer/__init__.py para implementar a extensão ao nível da aplicação:

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

Este código herda do AppExtensionBase para que a extensão se aplique a todas as funções na aplicação. Também poderia ter implementado a extensão num âmbito ao nível da função ao herdar do FuncExtensionBase.

O init método é um método de classe chamado pela função de trabalho quando a classe de extensão é importada. Pode efetuar ações de inicialização aqui para a extensão. Neste caso, é inicializado um mapa hash para registar a hora de início da invocação para cada função.

O configure método é orientado para o cliente. No seu ficheiro readme, pode informar os seus clientes quando precisarem de ligar Extension.configure()para . O readme também deve documentar as capacidades de extensão, a possível configuração e a utilização da sua extensão. Neste exemplo, os clientes podem escolher se o tempo decorrido é comunicado no HttpResponse.

O pre_invocation_app_level método é chamado pela função de trabalho python antes da execução da função. Fornece as informações da função, como o contexto da função e os argumentos. Neste exemplo, a extensão regista uma mensagem e regista a hora de início de uma invocação com base no respetivo invocation_id.

Da mesma forma, o é chamado após a post_invocation_app_level execução da função. Este exemplo calcula a hora decorrida com base na hora de início e na hora atual. Também substitui o valor devolvido da resposta HTTP.

Criar um readme.md

Crie um ficheiro readme.md na raiz do projeto de extensão. Este ficheiro contém as instruções e a utilização da sua extensão. O conteúdo readme.md é apresentado como a descrição na home page do projeto 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.

Consumir a extensão localmente

Agora que criou uma extensão, pode utilizá-la num projeto de aplicação para verificar se funciona conforme pretendido.

Criar uma função de acionador HTTP

Crie uma nova pasta para o seu projeto de aplicação e navegue para o mesmo.
A partir da shell adequada, como o Bash, execute o seguinte comando para inicializar o projeto:
```
func init --python
```
Utilize o seguinte comando para criar uma nova função de acionador HTTP que permita o acesso anónimo:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Ativar um ambiente virtual

Crie um ambiente virtual python, com base no SO da seguinte forma:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Ative o ambiente virtual do Python, com base no SO da seguinte forma:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Configurar a extensão

Instale pacotes remotos para o projeto da aplicação de funções com o seguinte comando:
```
pip install -r requirements.txt
```
Instale a extensão a partir do caminho do ficheiro local, no modo editável da seguinte forma:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
Neste exemplo, substitua <PYTHON_WORKER_EXTENSION_ROOT> pela localização do ficheiro de raiz do projeto de extensão.

Quando um cliente utiliza a sua extensão, em vez disso, adiciona a localização do pacote de extensão ao ficheiro de requirements.txt, tal como nos seguintes exemplos:
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
Abra o ficheiro de projeto local.settings.json e adicione o seguinte campo a Values:
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Ao executar no Azure, em vez disso, adiciona PYTHON_ENABLE_WORKER_EXTENSIONS=1 às definições da aplicação na aplicação de funções.
Adicione duas linhas seguintes antes da main função no ficheiro __init.py__ para o modelo de programação v1 ou no ficheiro function_app.py do modelo de programação v2:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Este código importa o TimerExtension módulo e define o valor de append_to_http_response configuração.

Verificar a extensão

A partir da pasta raiz do projeto de aplicação, inicie o anfitrião de funções com func host start --verbose. Deverá ver o ponto final local da sua função na saída como https://localhost:7071/api/HttpTrigger.

No browser, envie um pedido GET para https://localhost:7071/api/HttpTrigger. Deverá ver uma resposta como a seguinte, com os dados TimeElapsed do pedido acrescentados.

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)

Publicar a extensão

Depois de criar e verificar a extensão, ainda tem de concluir estas tarefas de publicação restantes:

Escolha uma licença.
Crie uma readme.md e outra documentação.
Publique a biblioteca de extensões num registo de pacotes python ou num sistema de controlo de versões (VCS).

PyPI
GitHub

Para publicar a extensão no PyPI:

Execute o seguinte comando para instalar twine e wheel no seu ambiente python predefinido ou num ambiente virtual:
```
pip install twine wheel
```
Remova a pasta antiga dist/ do repositório de extensões.
Execute o seguinte comando para gerar um novo pacote dentro de dist/:
```
python setup.py sdist bdist_wheel
```
Execute o seguinte comando para carregar o pacote para o PyPI:
```
twine upload dist/*
```
Poderá ter de fornecer as credenciais da conta PyPI durante o carregamento. Também pode testar o carregamento do pacote com twine upload -r testpypi dist/*. Para obter mais informações, veja a documentação do Twine.

Após estes passos, os clientes podem utilizar a sua extensão ao incluir o nome do pacote no respetivo requirements.txt.

Para obter mais informações, veja o tutorial oficial de empacotamento do Python.

Exemplos

Pode ver o projeto de extensão de exemplo concluído a partir deste artigo no python_worker_extension_timer repositório de exemplo.
A integração do OpenCensus é um projeto open source que utiliza a interface de extensão para integrar o rastreio de telemetria em aplicações Python Funções do Azure. Veja o repositório opencensus-python-extensions-azure para rever a implementação desta extensão de trabalho python.

Passos seguintes

Para obter mais informações sobre Funções do Azure desenvolvimento do Python, veja os seguintes recursos:

Partilhar via