Desenvolver extensões worker em Python para Azure Functions

Observação

A partir do Python 3.13, as extensões de 'workers' em Python deixarão de ser suportadas.

O Azure Functions permite integrar comportamentos personalizados como parte da execução de funções em Python. Esta funcionalidade permite-lhe criar lógica de negócio que os clientes podem usar facilmente nas suas próprias aplicações funcionais. Extensões worker são suportadas tanto nos modelos de programação Python v1 como v2.

Neste tutorial, você aprenderá a:

Crie uma extensão worker em Python ao nível da aplicação para o Azure Functions.
Consume a sua extensão numa aplicação da mesma forma que os seus clientes fazem.
Empacotar e publicar uma extensão para utilização.

Pré-requisitos

Antes de começar, deve cumprir estes requisitos:

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

Criar a extensão Python Worker

A extensão que crias reporta o tempo decorrido de uma invocação de trigger HTTP nos logs da consola e no corpo da resposta HTTP.

Estrutura de pastas

A pasta do seu projeto de extensão deve ser a 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 desenvolvimento local.
python_worker_extension/	Contém o código-fonte da extensão worker em Python. Esta pasta contém o módulo principal de Python a ser publicado no PyPI.
setup.py	Contém os metadados do pacote de extensão worker em Python.
readme.md	Contém as instruções e o uso da sua extensão. Este conteúdo é apresentado como a descrição na página inicial do seu projeto PyPI.

Configurar metadados do projeto

Primeiro cria setup.py, que fornece informações essenciais sobre o seu pacote. Para garantir que a sua extensão está distribuída e integrada corretamente nas funcionalidades das aplicações do cliente, confirme que 'azure-functions >= 1.7.0, < 2.0.0' está na secção install_requires.

No modelo seguinte, deve alterar author, author_email, install_requires, license, packages, e url campos 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,
)

De seguida, vai implementar o seu código de extensão no nível de aplicação.

Implementar a extensão do 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 da aplicação. Também podias ter implementado a extensão num âmbito ao nível da função, herdando do FuncExtensionBase.

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

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

O pre_invocation_app_level método é chamado pelo trabalhador Python antes da execução da função. Fornece a informação 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 na sua invocation_id.

De forma semelhante, o post_invocation_app_level é 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. Também sobrescrive o valor de retorno da resposta HTTP.

Crie um ficheiro readme.md

Cria um ficheiro readme.md na raiz do teu projeto de extensão. Este ficheiro contém as instruções e o uso da sua extensão. O conteúdo readme.md é apresentado como a descrição na página inicial do seu 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.

Utilize a sua extensão localmente

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

Criar uma função de gatilho HTTP

Crie uma nova pasta para o seu projeto de aplicação e navegue até ela.
A partir do shell apropriado, como o Bash, executa o seguinte comando para inicializar o projeto:
```
func init --python
```
Use o seguinte comando para criar uma nova função de gatilho HTTP que permita acesso anónimo:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Ativar um ambiente virtual

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

Configurar a extensão

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

Quando um cliente usa a sua extensão, ele irá adicionar a localização do seu pacote de extensões ao ficheiro requirements.txt, 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 do projeto local.settings.json e adicione o seguinte campo a Values:
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
Quando executares no Azure, deves adicionar PYTHON_ENABLE_WORKER_EXTENSIONS=1 às definições da aplicação na aplicação de funções.
Adicione as 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 para o modelo de programação v2:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
Este código importa o módulo TimerExtension e define o valor de configuração append_to_http_response.

Verificar a extensão

A partir da pasta raiz do projeto da tua app, inicia a função host usando func host start --verbose. Deverá ver o ponto final local da sua função na saída como https://localhost:7071/api/HttpTrigger.

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

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)

Publique sua extensão

Depois de criar e verificar a sua extensão, ainda precisa de completar estas tarefas de publicação restantes:

Escolha uma licença.
Crie um 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 sua extensão para o PyPI:

Execute o seguinte comando para instalar twine e wheel no seu ambiente Python padrão ou num ambiente virtual:
```
pip install twine wheel
```
Remove a pasta antiga dist/ do teu 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/*
```
Pode ser necessário fornecer as credenciais da sua conta PyPI durante o upload. Também pode testar o upload do seu pacote com twine upload -r testpypi dist/*. Para mais informações, consulte a documentação do Twine.

Após estes passos, os clientes podem usar a sua extensão incluindo o nome do seu pacote na requirements.txt.

Para mais informações, consulte o tutorial oficial de embalagem em Python.

Examples

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

Próximos passos

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

Feedback

Esta página foi útil?

Last updated on 2026-01-16