Desenvolva extensões de trabalho do Python para o Azure Functions

Artigo
04/29/2023

O Azure Functions permite integrar comportamentos personalizados como parte da execução da função do Python. Esse recurso permite que você crie uma lógica de negócios que os clientes poderão usar facilmente em seus próprios aplicativos de funções. Para saber mais, consulte a referência do desenvolvedor Python. Há suporte para extensões de trabalho nos modelos de programação Python v1 e v2.

Neste tutorial, você aprenderá como:

Crie uma extensão de trabalho do Python no nível do aplicativo para o Azure Functions.
Consuma sua extensão em um aplicativo da maneira que os clientes fazem.
Empacote e publique uma extensão para consumo.

Pré-requisitos

Antes de começar, será necessário atender a estes requisitos:

Python 3.7 ou superior. Para verificar a lista completa de versões do Python com suporte no Azure Functions, consulte o Guia do desenvolvedor do Python.
O Azure Functions Core Tools, versão 4.0.5095 ou posterior, que dá suporte ao uso da extensão com o modelo de programação do Python v2. Verifique sua versão com func --version.
Visual Studio Code instalado em uma das plataformas compatíveis.

Criar a extensão de trabalho do Python

A extensão que você cria relata o tempo decorrido de uma chamada de gatilho HTTP nos logs do console e no corpo da resposta HTTP.

Estrutura de pastas

A pasta para o projeto de extensão deve ter a seguinte estrutura:

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

Pasta/arquivo	Descrição
.venv/	(Opcional) Contém um ambiente virtual do Python usado para 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 ser publicado no PyPI.
setup.py	Contém os metadados do pacote de extensão de trabalho do Python.
readme.md	Contém as instruções e o uso da extensão. Este conteúdo é exibido como uma descrição na página inicial do projeto PyPI.

Configurar os metadados do projeto

Primeiro você cria setup.py, que fornece as informações essenciais sobre o pacote. Para verificar se a extensão está distribuída e integrada corretamente aos aplicativos de funções do cliente, confirme se 'azure-functions >= 1.7.0, < 2.0.0' está na seção install_requires.

No modelo a seguir, altere os campos author, author_email, install_requires, license, packages e 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, você implementará o código de extensão no escopo no nível de aplicativo.

Implementar a extensão do temporizador

Adicione o seguinte código em python_worker_extension_timer/__init__.py para implementar a extensão no nível de aplicativo:

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 é herdado de AppExtensionBase para que a extensão se aplique a todas as funções do aplicativo. Também seria possível implementar a extensão em um escopo no nível de função herdando de FuncExtensionBase.

O método init é um método de classe que é chamado pelo trabalho quando a classe de extensão é importada. É possível fazer ações de inicialização para a extensão. Nesse caso, um mapa de hash é inicializado para registrar a hora de início da chamada para cada função.

O método configure é voltado para o cliente. No arquivo leiame, é possível informar aos clientes quando a chamada Extension.configure() será necessária. O leiame também deve documentar os recursos de extensão, a configuração possível e o uso da extensão. Neste exemplo, os clientes podem escolher se o tempo decorrido será relatado em HttpResponse.

O método pre_invocation_app_level será chamado pelo trabalho do Python antes da função executar. Isso fornecerá as informações da função, como o contexto e os argumentos da função. Neste exemplo, a extensão registra uma mensagem e registra a hora de início de uma chamada com base no invocation_id.

Da mesma forma, post_invocation_app_level será chamado após a execução da função. Este exemplo calcula o tempo decorrido com base na hora de início e na hora atual. Ele também substitui o valor de retorno da resposta HTTP.

Criar um readme.md

Criar um arquivo readme.md na raiz do projeto de extensão. Esse arquivo contém as instruções e o uso da extensão. O conteúdo de readme.md é exibido como a descrição na página inicial 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 você criou uma extensão, poderá usá-la em um projeto de aplicativo para verificar se ela está funcionando como pretendido.

Criar uma função de gatilho HTTP

Crie uma nova pasta para o projeto de aplicativo e navegue até ela.
No shell apropriado, como Bash, execute o seguinte comando para inicializar o projeto:
```
func init --python
```
Use o seguinte comando para criar uma nova função de gatilho HTTP que permite acesso anônimo:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Ativar um ambiente virtual

Crie um ambiente virtual do Python baseado no sistema operacional da seguinte forma:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Ative o ambiente virtual do Python baseado no sistema operacional da seguinte maneira:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Configurar a extensão

Instale pacotes remotos para o projeto de aplicativo de funções usando o seguinte comando:
```
pip install -r requirements.txt
```
Instale a extensão do caminho do arquivo local, no modo editável, da seguinte maneira:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
Neste exemplo, substitua <PYTHON_WORKER_EXTENSION_ROOT> pelo local do arquivo raiz do seu projeto de extensão.

Quando um cliente usar a sua extensão, ele adicionará o local do pacote de extensão ao arquivo requirements.txt, como nos exemplos a seguir:
- 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 arquivo de projeto local.settings.json e adicione o seguinte campo para Values:
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Ao executar no Azure, você adiciona PYTHON_ENABLE_WORKER_EXTENSIONS=1 às configurações do aplicativo no aplicativo de funções.
Adicione as duas linhas a seguir antes da main função no arquivo __init.py__ para o modelo de programação v1 ou no arquivo function_app.py para o modelo de programação v2:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Esse código importa o módulo TimerExtension e define o valor da configuração append_to_http_response.

Verificar a extensão

Na pasta raiz do projeto do aplicativo, inicie o host de função usando func host start --verbose. Você deverá ver o ponto de extremidade local da função na saída como https://localhost:7071/api/HttpTrigger.
No navegador, envie uma solicitação GET para https://localhost:7071/api/HttpTrigger. Você deverá ver uma resposta semelhante à seguinte, com os dados TimeElapsed para a solicitação anexada.
```
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 sua extensão

Após criar e verificar a extensão, ainda será necessário concluir essas tarefas de publicação restantes:

Escolha um modelo de licença.
Crie uma readme.md e outra documentação.
Publique a biblioteca de extensão em um registro de pacote do Python ou em um VCS (sistema de controle de versão).

PyPI
GitHub

Para publicar a extensão no PyPI:

Execute o seguinte comando para instalar twine e wheel no ambiente padrão do Python ou em um ambiente virtual:
```
pip install twine wheel
```
Remova a pasta dist/ antiga do repositório de extensões.
Execute o seguinte comando para gerar um novo pacote em dist/:
```
python setup.py sdist bdist_wheel
```
Execute o seguinte comando para fazer upload do pacote para PyPI:
```
twine upload dist/*
```
Talvez você precise fornecer as credenciais da conta do PyPI durante o upload. Você também pode testar o upload do pacote com twine upload -r testpypi dist/*. Para obter mais informações, confira a documentação do Twine.

Após essas etapas, os clientes poderão usar sua extensão, incluindo o nome do pacote nos requisitos.txt.

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

Exemplos

Você pode exibir o projeto de extensão de amostra concluído neste artigo no repositório de amostra python_worker_extension_timer.
A integração do OpenCensus é um projeto de código aberto que usa a interface de extensão para integrar o rastreamento de telemetria em aplicativos Python do Azure Functions. Consulte o repositório opencensus-python-extensions-azure para revisar a implementação dessa extensão de trabalho do Python.

Próximas etapas

Para obter mais informações sobre o desenvolvimento do Python do Azure Functions consulte os seguintes recursos:

Share via