Desenvolva extensões de trabalho do Python para o Azure Functions
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:
Ative o ambiente virtual do Python baseado no sistema operacional da seguinte maneira:
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:
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çãoappend_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 comohttps://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).
Para publicar a extensão no PyPI:
Execute o seguinte comando para instalar
twine
ewheel
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: