Compartilhar via

Guia do desenvolvedor de Python para o Azure Functions

  • Artigo

Este guia é uma introdução ao desenvolvimento do Azure Functions usando o Python. O artigo considera que você já tenha lido o Guia de desenvolvedores do Azure Functions.

Importante

Este artigo dá suporte ao modelo de programação v1 e v2 para Python no Azure Functions. O modelo v1 do Python usa um arquivo functions.json para definir funções, enquanto o novo modelo v2 permite usar uma abordagem baseada em decorador. Essa nova abordagem resulta em uma estrutura de arquivos mais simples e mais centralizada em código. Escolha o seletor v2 na parte superior do artigo, para saber mais sobre esse novo modelo de programação.

Como desenvolvedor Python, você também pode estar interessado nesses tópicos:

  • Visual Studio Code: Crie seu primeiro aplicativo Python usando Visual Studio Code.
  • Terminal ou prompt de comando: Crie seu primeiro aplicativo Python no prompt de comando usando Azure Functions Core Tools.
  • Exemplos: Revise alguns aplicativos Python existentes no navegador de exemplos do Learn.
  • Visual Studio Code: Crie seu primeiro aplicativo Python usando Visual Studio Code.
  • Terminal ou prompt de comando: Crie seu primeiro aplicativo Python no prompt de comando usando Azure Functions Core Tools.
  • Exemplos: Revise alguns aplicativos Python existentes no navegador de exemplos do Learn.

Opções de desenvolvimento

Ambos os modelos de programação de funções do Python dão suporte ao desenvolvimento local em um dos seguintes ambientes:

Modelo de programação do Python v2:

Modelo de programação do Python v1:

Você também pode criar funções do Python v1 no portal do Azure.

Dica

Mesmo que seja possível desenvolver funções do Azure baseadas em Python localmente no Windows, o Python oferece suporte somente em um plano de hospedagem baseado em Linux quando ele é executado no Azure. Para obter mais informações, consulte a lista de combinações compatíveis de sistema operacional/runtime.

Modelo de programação

O Azure Functions espera que a função seja um método sem estado no script de Python que processa a entrada e produz a saída. Por padrão, o runtime espera que o método seja implementado como um método global chamado main() no arquivo __init__.py. Você também pode especificar um ponto de entrada alternativo.

Vincule dados às funções de gatilhos e associações por meio de atributos de método que usam a propriedade name, que é definida no arquivo function.json. Por exemplo, o arquivo function.json a seguir descreve uma função simples disparada por uma solicitação HTTP denominada req:

{
    "scriptFile": "__init__.py",
    "bindings": [
        {
            "authLevel": "function",
            "type": "httpTrigger",
            "direction": "in",
            "name": "req",
            "methods": [
                "get",
                "post"
            ]
        },
        {
            "type": "http",
            "direction": "out",
            "name": "$return"
        }
    ]
}

Com base nessa definição, o arquivo ___init__.py que contém o código de função pode ser semelhante a este exemplo:

def main(req):
    user = req.params.get('user')
    return f'Hello, {user}!'

Você também pode declarar explicitamente os tipos de atributo e de retorno na função usando anotações de tipo do Python. Realizar isso ajuda você a usar o IntelliSense e os recursos de preenchimento automático fornecidos por vários editores de código Python.

import azure.functions

def main(req: azure.functions.HttpRequest) -> str:
    user = req.params.get('user')
    return f'Hello, {user}!'

Use as anotações do Python incluídas no pacote azure.functions.* para associar a entrada e as saídas a seus métodos.

O Azure Functions espera que a função seja um método sem estado no script de Python que processa a entrada e produz a saída. Por padrão, o runtime espera que ela seja implementada como um método global no arquivo function_app.py.

Gatilhos e associações podem ser declarados e usados em uma função em uma abordagem baseada em decorador. Eles são definidos no mesmo arquivo, function_app.py, assim como as funções. Por exemplo, o arquivo function_app.py a seguir representa um gatilho de função por uma solicitação HTTP.

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req):
    user = req.params.get("user")
    return f"Hello, {user}!"

Você também pode declarar explicitamente os tipos de atributo e de retorno na função usando anotações de tipo do Python. Realizar isso ajuda você a usar o IntelliSense e os recursos de preenchimento automático fornecidos por vários editores de código Python.

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> str:
    user = req.params.get("user")
    return f"Hello, {user}!"

Para saber mais sobre limitações conhecidas com o modelo v2 e as soluções alternativas, confira Solucionar problemas de erros do Python no Azure Functions.

Ponto de entrada alternativo

Você poderá alterar o comportamento padrão de uma função se especificar opcionalmente as propriedades scriptFile e entryPoint no arquivo scriptFile. Por exemplo, o function.json a seguir diz ao runtime para usar o método customentry() no arquivo main.py como o ponto de entrada para sua função do Azure.

{
  "scriptFile": "main.py",
  "entryPoint": "customentry",
  "bindings": [
      ...
  ]
}

O ponto de entrada está apenas no arquivo function_app.py. No entanto, você pode fazer referência às funções dentro do projeto em function_app.py usando blueprints ou importando.

Estrutura de pastas

A estrutura de pastas recomendada a um projeto de funções do Python é semelhante a este exemplo:

 <project_root>/
 | - .venv/
 | - .vscode/
 | - my_first_function/
 | | - __init__.py
 | | - function.json
 | | - example.py
 | - my_second_function/
 | | - __init__.py
 | | - function.json
 | - shared_code/
 | | - __init__.py
 | | - my_first_helper_function.py
 | | - my_second_helper_function.py
 | - tests/
 | | - test_my_second_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

A principal pasta do projeto <project_root> pode conter os seguintes arquivos:

  • local.settings.json: usado para armazenar as configurações do aplicativo e as cadeias de conexão durante a execução local. Esse arquivo não é publicado no Azure. Para saber mais, confira local.settings.file.
  • requirements.txt: contém a lista de pacotes do Python que o sistema instala quando publica no Azure.
  • host.json: contém opções de configuração que afetam todas as funções em uma instância do aplicativo de funções. Esse arquivo é publicado do Azure. Nem todas as opções têm suporte quando executadas localmente. Para saber mais, confira host.json.
  • .vscode/: (opcional) contém a configuração do Visual Studio Code armazenadas. Para saber mais, confira Configurações de Visual Studio Code.
  • .venv/ : (opcional) contém um ambiente virtual do Python usado pelo desenvolvimento local.
  • Dockerfile: (opcional) usado quando o projeto é publicado em um contêiner personalizado.
  • tests/ : (opcional) contém os casos de teste do aplicativo de funções.
  • .funcignore: (opcional) declara os arquivos que não devem ser publicados no Azure. Normalmente, esse arquivo contém .vscode/ para ignorar a configuração do editor, .venv/ para ignorar o ambiente virtual do Python local, tests/ para ignorar casos de teste e local.settings.json para impedir que as configurações do aplicativo local sejam publicadas.

Cada função possui seu próprio arquivo de código e arquivo de configuração de associação function.json.

A estrutura de pastas recomendada a um projeto de funções do Python é semelhante a este exemplo:

 <project_root>/
 | - .venv/
 | - .vscode/
 | - function_app.py
 | - additional_functions.py
 | - tests/
 | | - test_my_function.py
 | - .funcignore
 | - host.json
 | - local.settings.json
 | - requirements.txt
 | - Dockerfile

A principal pasta do projeto <project_root> pode conter os seguintes arquivos:

  • .venv/: (opcional) contém um ambiente virtual do Python que é usado pelo desenvolvimento local.
  • .vscode/: (opcional) contém a configuração do Visual Studio Code armazenadas. Para saber mais, confira Configurações do Visual Studio Code.
  • function_app.py: o local padrão de todas as funções e seus respectivos gatilhos e associações.
  • additional_functions.py: (opcional) qualquer outro arquivo Python que contenha funções (geralmente para agrupamento lógico) que são referenciadas em function_app.py por meio de blueprints.
  • tests/ : (opcional) contém os casos de teste do aplicativo de funções.
  • .funcignore: (opcional) declara os arquivos que não devem ser publicados no Azure. Normalmente, esse arquivo contém .vscode/ para ignorar a configuração do editor, .venv/ para ignorar o ambiente virtual do Python local, tests/ para ignorar casos de teste e local.settings.json para impedir a publicação das configurações do aplicativo local.
  • host.json: contém opções de configuração que afetam todas as funções em uma instância do aplicativo de funções. Esse arquivo é publicado do Azure. Nem todas as opções têm suporte quando executadas localmente. Para saber mais, confira host.json.
  • local.settings.json: usado para armazenar as configurações do aplicativo e as cadeias de conexão ao ser executado localmente. Esse arquivo não é publicado no Azure. Para saber mais, confira local.settings.file.
  • requirements.txt: contém a lista de pacotes do Python que o sistema instala quando ele publica no Azure.
  • Dockerfile: (opcional) usado quando o projeto é publicado em um contêiner personalizado.

Quando você implanta o projeto em um aplicativo de funções no Azure, todo o conteúdo da principal pasta do projeto <project_root> deve ser incluído no pacote, mas não a pasta em si, o que significa que host.json deve estar na raiz do pacote. É recomendável manter os testes em uma pasta junto com outras funções (neste exemplo, é tests/). Para obter mais informações, confira Teste de unidade.

Conectar a um banco de dados

O Azure Functions integra-se bem ao Azure Cosmos DB para muitos casos de uso, incluindo IoT, comércio eletrônico, jogos etc.

Por exemplo, para fornecimento de eventos, os dois serviços são integrados para alimentar arquiteturas orientadas por eventos, usando a funcionalidade de feed de alterações do Azure Cosmos DB. O feed de alterações fornece a capacidade de ler inserções e atualizações (por exemplo, eventos de pedidos) de forma confiável e incremental, até o nível de microsserviços. Esta funcionalidade pode ser usada para fornecer um repositório de eventos persistente como um agente de mensagens para eventos de alteração de estado e fluxo de trabalho de processamento de pedidos de unidade entre muitos microsserviços (que podem ser implementados como Azure Functions sem servidor).

Arquitetura de referência de pipeline de ordenação do Azure Cosmos DB

Para se conectar ao Azure Cosmos DB, primeiro crie uma conta, um banco de dados e um contêiner. Em seguida, você pode conectar seu código de função ao Azure Cosmos DB usando gatilhos e associações, como este exemplo.

Para implementar uma lógica de aplicativo mais complexa, você também pode usar a biblioteca Python para Cosmos DB. Uma implementação de E/S assíncrona tem esta aparência:

pip install azure-cosmos
pip install aiohttp

from azure.cosmos.aio import CosmosClient
from azure.cosmos import exceptions
from azure.cosmos.partition_key import PartitionKey
import asyncio

# Replace these values with your Cosmos DB connection information
endpoint = "https://azure-cosmos-nosql.documents.azure.com:443/"
key = "master_key"
database_id = "cosmicwerx"
container_id = "cosmicontainer"
partition_key = "/partition_key"

# Set the total throughput (RU/s) for the database and container
database_throughput = 1000

# Singleton CosmosClient instance
client = CosmosClient(endpoint, credential=key)

# Helper function to get or create database and container
async def get_or_create_container(client, database_id, container_id, partition_key):
    database = await client.create_database_if_not_exists(id=database_id)
    print(f'Database "{database_id}" created or retrieved successfully.')

    container = await database.create_container_if_not_exists(id=container_id, partition_key=PartitionKey(path=partition_key))
    print(f'Container with id "{container_id}" created')
 
    return container
 
async def create_products():
    container = await get_or_create_container(client, database_id, container_id, partition_key)
    for i in range(10):
        await container.upsert_item({
            'id': f'item{i}',
            'productName': 'Widget',
            'productModel': f'Model {i}'
        })
 
async def get_products():
    items = []
    container = await get_or_create_container(client, database_id, container_id, partition_key)
    async for item in container.read_all_items():
        items.append(item)
    return items

async def query_products(product_name):
    container = await get_or_create_container(client, database_id, container_id, partition_key)
    query = f"SELECT * FROM c WHERE c.productName = '{product_name}'"
    items = []
    async for item in container.query_items(query=query, enable_cross_partition_query=True):
        items.append(item)
    return items

async def main():
    await create_products()
    all_products = await get_products()
    print('All Products:', all_products)

    queried_products = await query_products('Widget')
    print('Queried Products:', queried_products)

if __name__ == "__main__":
    asyncio.run(main())

Blueprints

O modelo de programação v2 do Python apresenta o conceito de blueprints. Um blueprint é uma nova classe que é instanciada para registrar funções fora do aplicativo de função principal. As funções registradas em instâncias de blueprint não são indexadas diretamente pelo runtime de função. Para indexar essas funções de blueprint, o aplicativo de funções precisa registrar as funções de instâncias de blueprint.

O uso de blueprints oferece os seguintes benefícios:

  • Permite dividir o aplicativo de funções em componentes modulares, que possibilita definir funções em vários arquivos Python e dividi-las em diferentes componentes por arquivo.
  • Fornece interfaces de aplicativos de funções públicas extensíveis para criar e reutilizar suas próprias APIs.

O exemplo a seguir mostra como usar os blueprints:

Primeiro, em um arquivo http_blueprint.py, uma função disparada por HTTP é definida pela primeira vez e adicionada a um objeto de blueprint.

import logging 

import azure.functions as func 

bp = func.Blueprint() 

@bp.route(route="default_template") 
def default_template(req: func.HttpRequest) -> func.HttpResponse: 
    logging.info('Python HTTP trigger function processed a request.') 

    name = req.params.get('name') 
    if not name: 
        try: 
            req_body = req.get_json() 
        except ValueError: 
            pass 
        else: 
            name = req_body.get('name') 

    if name: 
        return func.HttpResponse( 
            f"Hello, {name}. This HTTP-triggered function " 
            f"executed successfully.") 
    else: 
        return func.HttpResponse( 
            "This HTTP-triggered function executed successfully. " 
            "Pass a name in the query string or in the request body for a" 
            " personalized response.", 
            status_code=200 
        )

Em seguida, no arquivo function_app.py, o objeto de blueprint é importado e suas funções são registradas no aplicativo de funções.

import azure.functions as func 
from http_blueprint import bp

app = func.FunctionApp() 

app.register_functions(bp)

Observação

O Durable Functions também dá suporte a blueprints. Para criar blueprints para aplicativos de Durable Functions, registre sua orquestração, atividade, gatilhos de entidade e vinculações de cliente usando a classe azure-functions-durable Blueprint conforme mostrado aqui. O blueprint resultante pode ser registrado normalmente. Confira nossa amostra como exemplo.

Comportamento de importação

É possível importar módulos no código de função usando referências tanto absolutas quanto relativas. Com base na estrutura de pastas descrita anteriormente, as seguintes importações funcionam dentro do arquivo de função <project_root>\my_first_function\__init__.py:

from shared_code import my_first_helper_function #(absolute)

import shared_code.my_second_helper_function #(absolute)

from . import example #(relative)

Observação

Quando estiver usando a sintaxe de importação absoluta, a pasta shared_code/ precisa conter um arquivo __init__.py para marcá-lo como um pacote do Python.

A importação __app__ import e a importação relativa além do nível superior a seguir são preteridas, pois elas não têm suporte do verificador de tipo estático nem nas estruturas de teste do Python:

from __app__.shared_code import my_first_helper_function #(deprecated __app__ import)

from ..shared_code import my_first_helper_function #(deprecated beyond top-level relative import)

Gatilhos e entradas

As entradas são divididas em duas categorias no Azure Functions: entrada do gatilho e outra entrada. Embora elas sejam diferentes no arquivo function.json, o uso delas é idêntico no código do Python. As cadeias de conexão ou os segredos para fontes de entrada e gatilho são mapeados para valores no arquivo local.settings.json quando são executados localmente, e eles são mapeadas para as configurações do aplicativo quando executados no Azure.

Por exemplo, o código abaixo demonstra a diferença entre duas entradas:

// function.json
{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous",
      "route": "items/{id}"
    },
    {
      "name": "obj",
      "direction": "in",
      "type": "blob",
      "path": "samples/{id}",
      "connection": "STORAGE_CONNECTION_STRING"
    }
  ]
}

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>"
  }
}

# __init__.py
import azure.functions as func
import logging

def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

Quando a função é invocada, a solicitação HTTP é transmitida para a função como req. Uma entrada é recuperada da conta do Armazenamento de Blobs do Azure com base na ID na URL de rota e disponibilizada como obj no corpo da função. Aqui, a conta de armazenamento especificada é a cadeia de conexão encontrada na configuração CONNECTION_STRING do aplicativo.

As entradas são divididas em duas categorias no Azure Functions: entrada do gatilho e outra entrada. Embora sejam definidas usando decoradores diferentes, o uso delas é semelhante no código Python. As cadeias de conexão ou os segredos para fontes de entrada e gatilho são mapeados para valores no arquivo local.settings.json quando são executados localmente, e eles são mapeados para as configurações do aplicativo quando executados no Azure.

Por exemplo, o código a seguir demonstra como definir uma associação de entrada do Armazenamento de Blobs:

// local.settings.json
{
  "IsEncrypted": false,
  "Values": {
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "STORAGE_CONNECTION_STRING": "<AZURE_STORAGE_CONNECTION_STRING>",
    "AzureWebJobsStorage": "<azure-storage-connection-string>"
  }
}

# function_app.py
import azure.functions as func
import logging

app = func.FunctionApp()

@app.route(route="req")
@app.read_blob(arg_name="obj", path="samples/{id}", 
               connection="STORAGE_CONNECTION_STRING")
def main(req: func.HttpRequest, obj: func.InputStream):
    logging.info(f'Python HTTP-triggered function processed: {obj.read()}')

Quando a função é invocada, a solicitação HTTP é transmitida para a função como req. Uma entrada é recuperada da conta do Armazenamento de Blobs do Azure com base na ID na URL de rota e disponibilizada como obj no corpo da função. Aqui, a conta de armazenamento especificada é a cadeia de conexão encontrada na configuração STORAGE_CONNECTION_STRING do aplicativo.

Para as operações de associação com uso intensivo de dados, o ideal é usar uma conta de armazenamento separada. Para obter mais informações, confira Diretrizes de contas de armazenamento.

Associações de tipo do SDK (versão prévia)

Para selecionar gatilhos e associações, você pode trabalhar com tipos de dados implementados pelos SDKs e estruturas subjacentes do Azure. Essas associações de tipo de SDK permitem que você interaja dados de associação como se estivesse usando o SDK do serviço subjacente.

Importante

O suporte para associações de tipo SDK requer o modelo de programação Python v2.

O Functions dá suporte a associações de tipo do SDK do Python para o Armazenamento de Blobs do Azure, que permite trabalhar com dados de blob usando o tipo subjacente BlobClient.

Importante

O suporte a associações de tipo SDK para Python está atualmente em versão prévia:

  • Você deve usar o modelo de programação do Python v2.
  • Atualmente, há suporte apenas para tipos SDK síncronos.

Pré-requisitos

Habilitar associações de tipo SDK para a extensão de Armazenamento de Blobs

  1. Adicione o pacote de extensão azurefunctions-extensions-bindings-blob ao arquivo requirements.txt no projeto, que deve incluir pelo menos estes pacotes:

    azure-functions
azurefunctions-extensions-bindings-blob

  2. Adicione este código ao arquivo function_app.py no projeto, que importa as associações de tipo do SDK:

    import azurefunctions.extensions.bindings.blob as blob

Exemplos de associações de tipo SDK

Este exemplo mostra como obter BlobClient a partir de um gatilho de Armazenamento de Blobs (blob_trigger) e da associação de entrada em um gatilho HTTP (blob_input):

import logging

import azure.functions as func
import azurefunctions.extensions.bindings.blob as blob

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.blob_trigger(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_trigger(client: blob.BlobClient):
    logging.info(
        f"Python blob trigger function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )


@app.route(route="file")
@app.blob_input(
    arg_name="client", path="PATH/TO/BLOB", connection="AzureWebJobsStorage"
)
def blob_input(req: func.HttpRequest, client: blob.BlobClient):
    logging.info(
        f"Python blob input function processed blob \n"
        f"Properties: {client.get_blob_properties()}\n"
        f"Blob content head: {client.download_blob().read(size=1)}"
    )
    return "ok"

Você pode exibir outros exemplos de associações de tipo SDK para o Armazenamento de Blobs no repositório de extensões do Python:

Fluxos HTTP (versão prévia)

Os fluxos HTTP permitem que você aceite e retorne dados de seus pontos de extremidade HTTP usando APIs de solicitação e resposta FastAPI habilitadas em suas funções. Essas APIs permitem que o host processe dados grandes em mensagens HTTP como partes em vez de ler uma mensagem inteira na memória.

Este recurso possibilita lidar com grandes fluxos de dados, integrações do OpenAI, fornecer conteúdo dinâmico e dar suporte a outros cenários HTTP principais que exigem interações em tempo real por HTTP. Você também pode usar tipos de resposta FastAPI com fluxos HTTP. Sem fluxos HTTP, o tamanho de suas solicitações e respostas HTTP é limitado por restrições de memória que podem ser encontradas ao processar cargas de mensagens inteiras na memória.

Importante

O suporte para fluxos HTTP requer o modelo de programação Python v2.

Importante

O suporte a fluxos HTTP para Python está atualmente em versão prévia e exige que você use o modelo de programação do Python v2.

Pré-requisitos

Habilitar fluxos HTTP

Os fluxos HTTP estão desabilitados por padrão. Você precisa habilitar este recurso nas configurações do aplicativo e também atualizar o seu código para usar o pacote FastAPI. Observe que, ao habilitar os fluxos HTTP, o aplicativo de funções usará o streaming HTTP e a funcionalidade HTTP original não funcionará.

  1. Adicione o pacote de extensão azurefunctions-extensions-http-fastapi ao arquivo requirements.txt no projeto, que deve incluir pelo menos estes pacotes:

    azure-functions
azurefunctions-extensions-http-fastapi

  2. Adicione este código ao arquivo function_app.py no projeto, que importa a extensão FastAPI:

    from azurefunctions.extensions.http.fastapi import Request, StreamingResponse

  3. Ao implantar no Azure, adicione a seguinte configuração de aplicativo no seu aplicativo de funções:

    "PYTHON_ENABLE_INIT_INDEXING": "1"

    Se você estiver implantando no Consumo do Linux, também adicione

    "PYTHON_ISOLATE_WORKER_DEPENDENCIES": "1"

    Ao executar localmente, você também precisa adicionar estas mesmas configurações ao arquivo de projeto local.settings.json.

Exemplos de fluxos HTTP

Depois de habilitar o recurso de streaming HTTP, você pode criar funções que transmitem dados por HTTP.

Este exemplo é uma função disparada por HTTP que transmite dados de resposta HTTP. Você pode usar esses recursos para dar suporte a cenários como enviar dados de eventos por meio de um pipeline para visualização em tempo real ou detectar anomalias em grandes conjuntos de dados e fornecer notificações instantâneas.

import time

import azure.functions as func
from azurefunctions.extensions.http.fastapi import Request, StreamingResponse

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


def generate_sensor_data():
    """Generate real-time sensor data."""
    for i in range(10):
        # Simulate temperature and humidity readings
        temperature = 20 + i
        humidity = 50 + i
        yield f"data: {{'temperature': {temperature}, 'humidity': {humidity}}}\n\n"
        time.sleep(1)


@app.route(route="stream", methods=[func.HttpMethod.GET])
async def stream_sensor_data(req: Request) -> StreamingResponse:
    """Endpoint to stream real-time sensor data."""
    return StreamingResponse(generate_sensor_data(), media_type="text/event-stream")

Este exemplo é uma função disparada por HTTP que recebe e processa dados de streaming de um cliente em tempo real. Ele demonstra recursos de carregamento de streaming que podem ser úteis para cenários como o processamento de fluxos de dados contínuos e o tratamento de dados de eventos de dispositivos IoT.

import azure.functions as func
from azurefunctions.extensions.http.fastapi import JSONResponse, Request

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)


@app.route(route="streaming_upload", methods=[func.HttpMethod.POST])
async def streaming_upload(req: Request) -> JSONResponse:
    """Handle streaming upload requests."""
    # Process each chunk of data as it arrives
    async for chunk in req.stream():
        process_data_chunk(chunk)

    # Once all data is received, return a JSON response indicating successful processing
    return JSONResponse({"status": "Data uploaded and processed successfully"})


def process_data_chunk(chunk: bytes):
    """Process each data chunk."""
    # Add custom processing logic here
    pass

Chamar fluxos HTTP

Você deve usar uma biblioteca de clientes HTTP para fazer chamadas de streaming para os pontos de extremidade FastAPI de uma função. A ferramenta cliente ou o navegador que você está usando pode não dar suporte nativo ao streaming ou talvez retorne somente a primeira parte dos dados.

Você pode usar um script de cliente como este para enviar dados de streaming para um ponto de extremidade HTTP:

import httpx # Be sure to add 'httpx' to 'requirements.txt'
import asyncio

async def stream_generator(file_path):
    chunk_size = 2 * 1024  # Define your own chunk size
    with open(file_path, 'rb') as file:
        while chunk := file.read(chunk_size):
            yield chunk
            print(f"Sent chunk: {len(chunk)} bytes")

async def stream_to_server(url, file_path):
    timeout = httpx.Timeout(60.0, connect=60.0)
    async with httpx.AsyncClient(timeout=timeout) as client:
        response = await client.post(url, content=stream_generator(file_path))
        return response

async def stream_response(response):
    if response.status_code == 200:
        async for chunk in response.aiter_raw():
            print(f"Received chunk: {len(chunk)} bytes")
    else:
        print(f"Error: {response}")

async def main():
    print('helloworld')
    # Customize your streaming endpoint served from core tool in variable 'url' if different.
    url = 'http://localhost:7071/api/streaming_upload'
    file_path = r'<file path>'

    response = await stream_to_server(url, file_path)
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

outputs

A saída pode ser expressa em parâmetros de saída e em valores retornados. Se houver apenas uma saída, recomendamos usar o valor retornado. Para várias saídas, você deve usar parâmetros de saída.

Para usar o valor retornado de uma função como o valor de uma associação de saída, a propriedade name da associação deve ser definida como $return no arquivo function.json.

Para produzir várias saídas, use o método set() fornecido pela interface azure.functions.Out a fim de atribuir um valor para a associação. Por exemplo, a função a seguir pode enviar uma mensagem para uma fila e também retornar uma resposta HTTP.

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "req",
      "direction": "in",
      "type": "httpTrigger",
      "authLevel": "anonymous"
    },
    {
      "name": "msg",
      "direction": "out",
      "type": "queue",
      "queueName": "outqueue",
      "connection": "STORAGE_CONNECTION_STRING"
    },
    {
      "name": "$return",
      "direction": "out",
      "type": "http"
    }
  ]
}

import azure.functions as func

def main(req: func.HttpRequest,
         msg: func.Out[func.QueueMessage]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

A saída pode ser expressa em parâmetros de saída e em valores retornados. Se houver apenas uma saída, recomendamos usar o valor retornado. Para múltiplas saídas, você precisará usar parâmetros de saída.

Para produzir várias saídas, use o método set() fornecido pela interface azure.functions.Out a fim de atribuir um valor para a associação. Por exemplo, a função a seguir pode enviar uma mensagem para uma fila e também retornar uma resposta HTTP.

# function_app.py
import azure.functions as func

app = func.FunctionApp()

@app.write_blob(arg_name="msg", path="output-container/{name}",
                connection="CONNECTION_STRING")
def test_function(req: func.HttpRequest,
                  msg: func.Out[str]) -> str:

    message = req.params.get('body')
    msg.set(message)
    return message

Registro em log

O acesso ao agente do Azure Functions runtime está disponível por meio de um manipulador logging raiz no seu aplicativo de funções. Esse agente é vinculado ao Application Insights e permite sinalizar avisos e erros que ocorrem durante a execução da função.

O exemplo a seguir registra uma mensagem de informações quando a função é invocada por meio de um gatilho HTTP.

import logging

def main(req):
    logging.info('Python HTTP trigger function processed a request.')

Mais métodos de registro em log disponíveis que permitem a gravação no console em níveis de rastreamento diferentes:

Método Descrição
critical(_message_) Grava uma mensagem com nível CRÍTICO no agente raiz.
error(_message_) Grava uma mensagem com nível ERRO no agente raiz.
warning(_message_) Grava uma mensagem com nível AVISO no agente raiz.
info(_message_) Grava uma mensagem com nível INFORMAÇÕES no agente raiz.
debug(_message_) Grava uma mensagem com nível DEPURAR no agente raiz.

Para saber mais sobre como registrar em log, confira Monitorar o Azure Functions.

Registro em log de threads criados

Para ver os logs provenientes dos threads criados, inclua o argumento context na assinatura de função. Este argumento contém um atributo thread_local_storage que armazena um invocation_id local. Isso pode ser definido como o atual da função invocation_id para garantir que o contexto seja alterado.

import azure.functions as func
import logging
import threading


def main(req, context):
    logging.info('Python HTTP trigger function processed a request.')
    t = threading.Thread(target=log_function, args=(context,))
    t.start()


def log_function(context):
    context.thread_local_storage.invocation_id = context.invocation_id
    logging.info('Logging from thread.')

Registrar telemetria personalizada

Por padrão, o runtime do Functions coleta logs e outros dados de telemetria que são gerados por suas funções. Essa telemetria acaba como rastreamentos no Application Insights. Por padrão, também são coletadas telemetrias de solicitação e dependência de determinados serviços do Azure por meio de gatilhos e associações.

Para coletar telemetrias personalizadas de solicitação e dependência sem associações, use as Extensões OpenCensus do Python. Essa extensão envia dados de telemetria personalizados para sua instância de Application Insights. Você pode encontrar uma lista de extensões com suporte no repositório do OpenCensus.

Observação

Para usar as extensões OpenCensus do Python, é necessário habilitar as extensões de trabalho do Python no aplicativo de funções definindo PYTHON_ENABLE_WORKER_EXTENSIONS como 1. Você também precisará passar a usar a cadeia de conexão do Application Insights adicionando a configuração APPLICATIONINSIGHTS_CONNECTION_STRING às suas APPLICATIONINSIGHTS_CONNECTION_STRING, se ela ainda não estiver lá.

// requirements.txt
...
opencensus-extension-azure-functions
opencensus-ext-requests

import json
import logging

import requests
from opencensus.extension.azure.functions import OpenCensusExtension
from opencensus.trace import config_integration

config_integration.trace_integrations(['requests'])

OpenCensusExtension.configure()

def main(req, context):
    logging.info('Executing HttpTrigger with OpenCensus extension')

    # You must use context.tracer to create spans
    with context.tracer.span("parent"):
        response = requests.get(url='http://example.com')

    return json.dumps({
        'method': req.method,
        'response': response.status_code,
        'ctx_func_name': context.function_name,
        'ctx_func_dir': context.function_directory,
        'ctx_invocation_id': context.invocation_id,
        'ctx_trace_context_Traceparent': context.trace_context.Traceparent,
        'ctx_trace_context_Tracestate': context.trace_context.Tracestate,
        'ctx_retry_context_RetryCount': context.retry_context.retry_count,
        'ctx_retry_context_MaxRetryCount': context.retry_context.max_retry_count,
    })

Gatilho HTTP

O gatilho HTTP é definido no arquivo function.json. O name da associação deve corresponder ao parâmetro nomeado na função. Nos exemplos anteriores, um nome de associação req é usado. Esse parâmetro é um objeto HttpRequest, e um objeto HttpResponse é retornado.

No objeto HttpRequest, você pode obter cabeçalhos de solicitação, parâmetros de consulta, parâmetros de rota e o corpo da mensagem.

O exemplo a seguir é do modelo de gatilho HTTP para Python.

def main(req: func.HttpRequest) -> func.HttpResponse:
    headers = {"my-http-header": "some-value"}

    name = req.params.get('name')
    if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

    if name:
        return func.HttpResponse(f"Hello {name}!", headers=headers)
    else:
        return func.HttpResponse(
             "Please pass a name on the query string or in the request body",
             headers=headers, status_code=400
        )

Nessa função, você obtém o valor do parâmetro de consulta name a partir do parâmetro params do objeto HttpRequest. Você faz a leitura do corpo da mensagem codificada em JSON usando o método get_json.

Da mesma forma, você pode definir status_code e headers para a mensagem de resposta no objeto retornado status_code.

O gatilho HTTP é definido como um método que usa um parâmetro de associação nomeado, que é um objeto HttpRequest e retorna um objeto HttpResponse. Você aplica o decorador function_name ao método para definir o nome da função, enquanto o ponto de extremidade HTTP é definido aplicando o decorador route.

Este exemplo é do modelo de gatilho HTTP para o modelo de programação Python v2, em que o nome do parâmetro de associação é req. É o código de exemplo fornecido quando você cria uma função usando o Azure Functions Core Tools ou o Visual Studio Code.

@app.function_name(name="HttpTrigger1")
@app.route(route="hello")
def test_function(req: func.HttpRequest) -> func.HttpResponse:
     logging.info('Python HTTP trigger function processed a request.')

     name = req.params.get('name')
     if not name:
        try:
            req_body = req.get_json()
        except ValueError:
            pass
        else:
            name = req_body.get('name')

     if name:
        return func.HttpResponse(f"Hello, {name}. This HTTP-triggered function executed successfully.")
     else:
        return func.HttpResponse(
             "This HTTP-triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response.",
             status_code=200
        )

No objeto HttpRequest, você pode obter cabeçalhos de solicitação, parâmetros de consulta, parâmetros de rota e o corpo da mensagem. Nessa função, você obtém o valor do parâmetro de consulta name a partir do parâmetro params do objeto HttpRequest. Você faz a leitura do corpo da mensagem codificada em JSON usando o método get_json.

Da mesma forma, você pode definir status_code e headers para a mensagem de resposta no objeto retornado status_code.

Para passar um nome neste exemplo, cole a URL que foi fornecida ao executar a função e depois acrescente-a com "?name={name}".

Estruturas Web

Você pode usar estruturas compatíveis com WSGI (Interface de Gateway de Servidor Web) e compatíveis com ASGI (Interface de Gateway de Servidor Assíncrono), como Flask e FastAPI, com suas funções do Python disparadas por HTTP. Esta seção mostra como modificar suas funções para dar suporte a essas estruturas.

Primeiro, o arquivo function.json precisa ser atualizado para incluir um route no gatilho HTTP, conforme mostrado no seguinte exemplo:

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
       "authLevel": "anonymous",
       "type": "httpTrigger",
       "direction": "in",
       "name": "req",
       "methods": [
           "get",
           "post"
       ],
       "route": "{*route}"
    },
    {
       "type": "http",
       "direction": "out",
       "name": "$return"
    }
  ]
}

O arquivo host.json precisa ser atualizado para incluir um HTTP routePrefix, conforme mostrado no exemplo a seguir:

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[3.*, 4.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

Atualize o arquivo de código Python init.py, dependendo da interface que é usada pela sua estrutura. O seguinte exemplo mostra uma abordagem de manipulador de ASGI ou uma abordagem de wrapper WSGI para Flask:

app = fastapi.FastAPI()

@app.get("hello/{name}")
async def get_name(name: str):
  return {"name": name}

def main(req: func.HttpRequest, context: func.Context) -> func.HttpResponse:
    return func.AsgiMiddleware(app).handle(req, context)

Você pode usar estruturas compatíveis com ASGI (Interface de Gateway de Servidor Assíncrono) e com WSGI (Interface de Gateway de Servidor Web), como Flask e FastAPI, com suas funções do Python disparadas por HTTP. Você deve atualizar primeiro o arquivo host.json para incluir um HTTP routePrefix, conforme mostrado no seguinte exemplo:

{
  "version": "2.0",
  "logging": 
  {
    "applicationInsights": 
    {
      "samplingSettings": 
      {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": 
  {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[2.*, 3.0.0)"
  },
  "extensions": 
  {
    "http": 
    {
        "routePrefix": ""
    }
  }
}

Agora o código de estrutura deve ser como o seguinte exemplo:

AsgiFunctionApp é a classe de aplicativo de funções de nível superior para criar funções HTTP ASGI.

# function_app.py

import azure.functions as func 
from fastapi import FastAPI, Request, Response 

fast_app = FastAPI() 

@fast_app.get("/return_http_no_body") 
async def return_http_no_body(): 
    return Response(content="", media_type="text/plain") 

app = func.AsgiFunctionApp(app=fast_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS)

Dimensionamento e desempenho

Para obter as melhores práticas de dimensionamento e desempenho para aplicativos de funções do Python, confira o artigo Escala e desempenho no Python.

Contexto

Para obter o contexto de invocação de uma função quando está em execução, inclua o argumento context em sua assinatura.

Por exemplo:

import azure.functions


def main(req: azure.functions.HttpRequest,
         context: azure.functions.Context) -> str:
    return f'{context.invocation_id}'

A classeContext tem os seguintes atributos de cadeia de caracteres:

Atributo Descrição
function_directory O diretório no qual a função está em execução.
function_name O nome da função.
invocation_id A ID da invocação de função atual.
thread_local_storage O armazenamento local de thread da função. Contém um local invocation_id para registro em log de threads criados.
trace_context O contexto para rastreamento distribuído. Para obter mais informações, consulte Trace Context.
retry_context O contexto para novas tentativas para a função. Para obter mais informações, consulte retry-policies.

Variáveis globais

Não há garantia de que o estado do seu aplicativo será preservado para execuções futuras. No entanto, o runtime do Azure Functions geralmente reutiliza o mesmo processo para várias execuções do mesmo aplicativo. Declare uma computação cara como uma variável global para armazenar os resultados dela em cache.

CACHED_DATA = None


def main(req):
    global CACHED_DATA
    if CACHED_DATA is None:
        CACHED_DATA = load_json()

    # ... use CACHED_DATA in code

Variáveis de ambiente

No Azure Functions, as configurações de aplicativo, como cadeias de conexão de serviço, são expostas como variáveis de ambiente quando estão em execução. Há duas maneiras principais de acessar essas configurações em seu código.

Método Descrição
os.environ["myAppSetting"] Tenta obter a configuração de aplicativo por nome de chave e gera um erro quando ele é malsucedido.
os.getenv("myAppSetting") Tenta obter a configuração de aplicativo por nome de chave e retorna null quando ele é malsucedido.

Essas duas maneiras exigem que você declare import os.

O exemplo a seguir usa os.environ["myAppSetting"]para obter a os.environ["myAppSetting"], com a chave chamada myAppSetting:

import logging
import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

Para o desenvolvimento local, as configurações de aplicativo são mantidas no arquivo local.settings.json.

No Azure Functions, as configurações de aplicativo, como cadeias de conexão de serviço, são expostas como variáveis de ambiente quando estão em execução. Há duas maneiras principais de acessar essas configurações em seu código.

Método Descrição
os.environ["myAppSetting"] Tenta obter a configuração de aplicativo por nome de chave e gera um erro quando ele é malsucedido.
os.getenv("myAppSetting") Tenta obter a configuração de aplicativo por nome de chave e retorna null quando ele é malsucedido.

Essas duas maneiras exigem que você declare import os.

O exemplo a seguir usa os.environ["myAppSetting"]para obter a os.environ["myAppSetting"], com a chave chamada myAppSetting:

import logging
import os

import azure.functions as func

app = func.FunctionApp()

@app.function_name(name="HttpTrigger1")
@app.route(route="req")
def main(req: func.HttpRequest) -> func.HttpResponse:
  # Get the setting named 'myAppSetting'
  my_app_setting_value = os.environ["myAppSetting"]
  logging.info(f'My app setting value:{my_app_setting_value}')

Para o desenvolvimento local, as configurações de aplicativo são mantidas no arquivo local.settings.json.

Versão do Python

O Azure Functions dá suporte às seguintes versões do Python:

Versão do Functions Versões do Python*
4.x 3.11
3.10
3.9
3.8
3.7
3.x 3.9
3.8
3.7

* Distribuições oficiais do Python

Para solicitar uma versão específica do Python quando você cria seu aplicativo de funções no Azure, use a opção --runtime-version do comando az functionapp create. A versão de runtime do Functions é definida pela opção --functions-version. A versão do Python é definida quando o aplicativo de funções é criado e ela não pode ser alterada por aplicativos executados em um plano de Consumo.

O runtime usa a versão do Python disponível quando você o executa localmente.

Alterar a versão do Python

Para configurar um aplicativo de funções do Python para uma versão de linguagem específica, será necessário especificar a linguagem e a versão da linguagem no campo LinuxFxVersion na configuração do site. Por exemplo, para alterar o aplicativo do Python para usar o Python 3.8, defina linuxFxVersion como python|3.8.

Para saber como exibir e alterar a configuração do site linuxFxVersion, confira Como direcionar as versões de runtime do Azure Functions.

Para obter mais informações gerais, confira a política de suporte de runtime do Azure Functions e Idiomas com suporte no Azure Functions.

Gerenciamento de pacotes

Quando estiver desenvolvendo localmente usando o Core Tools ou o Visual Studio Code, adicione os nomes e as versões dos pacotes necessários para os arquivos requirements.txt e depois instale-os usando pip.

Por exemplo, é possível usar o arquivo requirements.txt e comando pip a seguir para instalar o pacote requests do PyPI.

requests==2.19.1

pip install -r requirements.txt

Ao executar suas funções em um plano do Serviço de Aplicativo, as dependências definidas em requirements.txt recebem precedência sobre módulos Python internos, como logging. Essa precedência pode causar conflitos quando módulos internos têm os mesmos nomes que diretórios em seu código. Ao executar em um plano de consumo ou em um plano Elastic Premium, os conflitos são menos prováveis porque suas dependências não são priorizadas por padrão.

Para evitar problemas em execução em um plano do Serviço de Aplicativo, não nomeie seus diretórios da mesma forma que os módulos nativos do Python e não inclua bibliotecas nativas do Python no arquivo requirements.txt do projeto.

Publicando no Azure

Quando tudo estiver pronto para publicar, verifique se todas as dependências disponíveis publicamente estão listadas no arquivo requirements.txt. Você pode localizar esse arquivo na raiz do diretório do projeto.

Você pode encontrar os arquivos e as pastas do projeto que foram excluídos da publicação, incluindo a pasta de ambiente virtual, no diretório raiz do projeto.

Existem três ações de build com suporte para publicar o projeto do Python no Azure: build remoto, build local e builds usando dependências personalizadas.

Também é possível usar o Azure Pipelines para criar dependências e publicar usando a CD (entrega contínua). Para saber mais, confira Configurar entrega com o Azure Pipelines.

Build remoto

No build remoto, as dependências que são restauradas no servidor e em dependências nativas correspondem ao ambiente de produção. Isso resulta em um pacote de implantação menor para upload. Usar compilação remota ao desenvolver aplicativos do Python no Windows. Se o projeto tiver dependências personalizadas, você poderá usar o build remoto com a URL de índice extra.

As dependências são obtidas remotamente com base no conteúdo do arquivo requirements.txt. O Build remoto é o método de build recomendado. Por padrão, o Core Tools solicita um build remoto quando você usa o comando func azure functionapp publish a seguir para publicar seu projeto Python no Azure.

func azure functionapp publish <APP_NAME>

Lembre-se de substituir <APP_NAME> pelo nome do aplicativo de funções no Azure.

A extensão do Azure Functions para Visual Studio Code também solicita um build remoto por padrão.

Build local

As dependências são obtidas localmente com base no conteúdo do arquivo requirements.txt. Você pode evitar fazer um build remoto com o uso do comando func azure functionapp publish a seguir para publicar com um build local:

func azure functionapp publish <APP_NAME> --build local

Lembre-se de substituir <APP_NAME> pelo nome do aplicativo de funções no Azure.

Com a opção --build local, as dependências do projeto são lidas no arquivo requirements.txt e esses pacotes dependentes são baixados e instalados localmente. Os arquivos e as dependências do projeto são implantados de seu computador local no Azure. Isso faz com que um pacote de implantação maior seja carregado no Azure. Se por algum motivo você não conseguir acessar o arquivo requirements.txt usando o Core Tools, é necessário usar a opção de dependências personalizadas para publicação.

Não é recomendável usar compilações locais ao desenvolver localmente no Windows.

Dependências personalizadas

Quando o projeto tem dependências que não são encontradas no Índice de Pacote do Python, há duas formas de compilar o projeto. A primeira forma, o método de build, depende de como o projeto é compilado.

Build remoto com URL de índice extra

Use um build remoto quando os pacotes estiverem disponíveis em um índice de pacote personalizado acessível. Antes de publicar, não deixe de criar uma configuração de aplicativo chamada PIP_EXTRA_INDEX_URL. O valor dessa configuração é a URL do índice de pacote personalizado. Essa configuração instrui o build remoto a executar pip install usando a opção --extra-index-url. Para saber mais, confira a Documentação pip installdo Python.

Também é possível usar credenciais básicas de autenticação com as URLs de índice de pacote extra. Para saber mais, confira Credenciais de autenticação básicas na documentação do Python.

Instalar pacotes locais

Se seu projeto usa pacotes que não estão disponíveis publicamente para nossas ferramentas, você pode disponibilizá-los para seu aplicativo se os colocar no diretório __app__/.python_packages. Antes de publicar, execute o seguinte comando para instalar as dependências localmente:

pip install  --target="<PROJECT_DIR>/.python_packages/lib/site-packages"  -r requirements.txt

Ao usar dependências personalizadas, você deve usar a opção de publicação --no-build, porque você já instalou as dependências na pasta do projeto.

func azure functionapp publish <APP_NAME> --no-build

Lembre-se de substituir <APP_NAME> pelo nome do aplicativo de funções no Azure.

Teste de unidade

As funções gravadas no Python podem ser testadas como outros códigos Python por meio de estruturas de teste padrão. Para a maioria das associações, é possível criar um objeto de entrada fictício com a criação de uma instância de uma classe apropriada do pacote azure.functions. Como o pacote azure.functions não fica disponível imediatamente, certifique-se de instalá-lo por meio do arquivo requirements.txt conforme descrito na seção Gerenciamento de pacotes acima.

Com my_second_function como exemplo, confira a seguir um teste fictício de uma função disparada por HTTP:

Primeiro, crie o arquivo <project_root>/my_second_function/function.json e defina essa função como um gatilho HTTP.

{
  "scriptFile": "__init__.py",
  "entryPoint": "main",
  "bindings": [
    {
      "authLevel": "function",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

Em seguida, você pode implementar my_second_function e shared_code.my_second_helper_function.

# <project_root>/my_second_function/__init__.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

# Define an HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
def main(req: func.HttpRequest) -> func.HttpResponse:
  logging.info('Executing my_second_function.')

  initial_value: int = int(req.params.get('value'))
  doubled_value: int = my_second_helper_function.double(initial_value)

  return func.HttpResponse(
    body=f"{initial_value} * 2 = {doubled_value}",
    status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

Comece gravando casos de teste para seu gatilho HTTP.

# <project_root>/tests/test_my_second_function.py
import unittest

import azure.functions as func
from my_second_function import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    resp = main(req)

    # Check the output.
    self.assertEqual(resp.get_body(), b'21 * 2 = 42',)

Dentro da pasta de ambiente virtual do Python .venv, instale sua estrutura de teste favorita do Python, como pip install pytest. Depois, execute pytest tests para verificar o resultado do teste.

Primeiro, crie o arquivo <project_root>/function_app.py e implemente a função my_second_function como o gatilho HTTP e shared_code.my_second_helper_function.

# <project_root>/function_app.py
import azure.functions as func
import logging

# Use absolute import to resolve shared_code modules
from shared_code import my_second_helper_function

app = func.FunctionApp()

# Define the HTTP trigger that accepts the ?value=<int> query parameter
# Double the value and return the result in HttpResponse
@app.function_name(name="my_second_function")
@app.route(route="hello")
def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Executing my_second_function.')

    initial_value: int = int(req.params.get('value'))
    doubled_value: int = my_second_helper_function.double(initial_value)

    return func.HttpResponse(
        body=f"{initial_value} * 2 = {doubled_value}",
        status_code=200
    )

# <project_root>/shared_code/__init__.py
# Empty __init__.py file marks shared_code folder as a Python package

# <project_root>/shared_code/my_second_helper_function.py

def double(value: int) -> int:
  return value * 2

Comece gravando casos de teste para seu gatilho HTTP.

# <project_root>/tests/test_my_second_function.py
import unittest
import azure.functions as func

from function_app import main

class TestFunction(unittest.TestCase):
  def test_my_second_function(self):
    # Construct a mock HTTP request.
    req = func.HttpRequest(method='GET',
                           body=None,
                           url='/api/my_second_function',
                           params={'value': '21'})
    # Call the function.
    func_call = main.build().get_user_function()
    resp = func_call(req)
    # Check the output.
    self.assertEqual(
        resp.get_body(),
        b'21 * 2 = 42',
    )

Dentro da pasta de ambiente virtual do Python .venv, instale sua estrutura de teste do Python favorita, como pip install pytest. Depois, execute pytest tests para verificar o resultado do teste.

Arquivos temporários

O método tempfile.gettempdir() retorna uma pasta temporária, que no Linux é a /tmp. Seu aplicativo pode usar esse diretório para armazenar arquivos temporários gerados e usados por suas funções quando estão em execução.

Importante

Os arquivos gravados no diretório temporário podem não persistir entre invocações. Durante o processo de escalar horizontalmente, os arquivos temporários não são compartilhados entre instâncias.

O exemplo abaixo cria um arquivo temporário nomeado no diretório temporário (/tmp):

import logging
import azure.functions as func
import tempfile

from os import listdir

#---
   tempFilePath = tempfile.gettempdir()
   fp = tempfile.NamedTemporaryFile()
   fp.write(b'Hello world!')
   filesDirListInTemp = listdir(tempFilePath)

Recomendamos que você mantenha os testes em uma pasta que é separada da pasta do projeto. Essa ação impede que você implante o código de teste com seu aplicativo.

Bibliotecas pré-instaladas

Algumas bibliotecas são fornecidas com o runtime de funções do Python.

A biblioteca padrão do Python

A biblioteca padrão do Python contém uma lista de módulos internos do Python que são fornecidos com cada distribuição do Python. A maioria dessas bibliotecas ajuda a acessar a funcionalidade do sistema, como a E/S (entrada/saída) de arquivo. Nos sistemas Windows, essas bibliotecas são instaladas com o Python. Em sistemas baseados em UNIX, elas são fornecidas por coleções de pacotes.

Para exibir a biblioteca para sua versão do Python, acesse:

Dependências de trabalho do Python no Azure Functions

O trabalho do Python no Azure Functions necessita de um conjunto específico de bibliotecas. Embora você também possa usar essas bibliotecas nas funções, elas não fazem parte do padrão do Python. Caso as funções dependam de alguma dessas bibliotecas, elas podem estar indisponíveis para o código quando elas forem executadas fora do Azure Functions.

Observação

Se o arquivo requirements.txt no aplicativo de funções contiver uma entrada azure-functions-worker, remova-a. O trabalho de funções é gerenciado automaticamente pela plataforma do Azure Functions e nós o atualizamos regularmente com novos recursos e correções de bugs. A instalação manual de uma versão antiga do trabalho no arquivo requirements.txt pode causar problemas inesperados.

Observação

Se o pacote contém determinadas bibliotecas que podem colidir com dependências de trabalho (por exemplo, protobuf, tensorflow ou grpcio), configure PYTHON_ISOLATE_WORKER_DEPENDENCIES como 1 nas configurações do aplicativo para evitar que o aplicativo faça referência à dependências de trabalho.

A biblioteca do Python no Azure Functions

Cada atualização de trabalho do Python inclui uma nova versão da Biblioteca do Python no Azure Functions (azure.functions). Essa abordagem facilita a atualização contínua dos aplicativos de funções do Python, pois cada atualização é compatível com versões anteriores. Para obter uma lista de versões dessa biblioteca, azure-functions PyPi.

A versão da biblioteca no runtime é corrigida pelo Azure e não pode ser substituída pelo arquivo requirements.txt. A entrada azure-functions no arquivo requirements.txt é apenas para lint e conscientização do cliente.

Use o código a seguir para rastrear a versão real da biblioteca de funções do Python no runtime:

getattr(azure.functions, '__version__', '< 1.2.1')

Bibliotecas do sistema no runtime

Para obter uma lista de bibliotecas do sistema pré-instaladas em imagens Docker de trabalho do Python, confira os links a seguir:

runtime do Functions Versão do Debian Versões do Python
Versão 3.x Buster Python 3.7
Python 3.8
Python 3.9

Extensões de trabalho do Python

O processo de trabalho do Python executado no Azure Functions permite integrar bibliotecas de terceiros ao aplicativo de funções. Essas bibliotecas de extensão atuam como middleware que pode injetar operações específicas durante o ciclo de vida da execução da função.

As extensões são importadas no código de função, assim como um módulo de biblioteca Python padrão. As extensões são baseadas em execução nos seguintes escopos:

Escopo Descrição
Nível de aplicativo Quando importado para qualquer gatilho de função, a extensão se aplica a cada execução de função no aplicativo.
Nível de função A execução é limitada apenas ao gatilho de função específico ao qual ela é importada.

Examine as informações de cada extensão para saber mais sobre o escopo no qual a extensão é executada.

As extensões implementam uma interface de extensão de trabalho do Python. Essa ação permite que o processo de trabalho do Python chame o código de extensão durante o ciclo de vida da execução de função. Para saber mais, confira Criar extensões.

Como usar extensões

Você pode usar uma biblioteca de extensões de trabalho do Python em suas funções do Python da seguinte maneira:

  1. Adicione o pacote de extensão no arquivo requirements.txt para o projeto.
  2. Instale a biblioteca no aplicativo.
  3. Adicione as seguintes configurações de aplicativo:
  4. Importe o módulo de extensão para o gatilho de função.
  5. Se necessário, configure a instância de extensão. Os requisitos de configuração devem ser chamados na documentação da extensão.

Importante

Não há suporte para bibliotecas de extensão de trabalho do Python de terceiros ou garantida pela Microsoft. Você deve certificar-se de que todas as extensões usadas no aplicativo de funções são confiáveis e que você assume o risco de usar uma extensão mal-intencionada ou gravada de forma precária.

Terceiros devem fornecer documentação específica sobre como instalar e consumir suas extensões no seu aplicativo de funções. Para obter um exemplo básico de como consumir uma extensão, confira Consumo da extensão.

Estes são alguns exemplos de como usar extensões em um aplicativo de funções, por escopo:

# <project_root>/requirements.txt
application-level-extension==1.0.0

# <project_root>/Trigger/__init__.py

from application_level_extension import AppExtension
AppExtension.configure(key=value)

def main(req, context):
  # Use context.app_ext_attributes here

Criar extensões

As extensões são criadas por desenvolvedores de biblioteca de terceiros que criaram funcionalidades que podem ser integradas ao Azure Functions. Um desenvolvedor de extensão projeta, implementa e libera pacotes do Python que contêm lógica personalizada projetada especificamente para ser executada no contexto da execução da função. Essas extensões podem ser publicadas no registro PyPI ou em repositórios GitHub.

Para saber como criar, empacotar, publicar e consumir um pacote de extensão de trabalho do Python, confira Desenvolver extensões de trabalho do Python para o Azure Functions.

Extensões de nível de aplicativo

Uma extensão que é herdada de AppExtensionBase é executada em um escopo de aplicativo.

AppExtensionBase expõe os seguintes métodos de classe abstrata para implementar:

Método Descrição
init Chamado depois que a extensão é importada.
configure Chamado do código de função quando é necessário para configurar a extensão.
post_function_load_app_level Chamado logo após a função ser carregada. O nome da função e o diretório de função são passados para a extensão. Não esqueça que o diretório de funções é somente leitura e toda tentativa de gravar no arquivo local falhará nesse diretório.
pre_invocation_app_level Chamado logo antes da função ser disparada. Os argumentos de invocação de função e de contexto de função são passados para a extensão. Geralmente, você pode passar outros atributos no objeto de contexto para serem consumidos pelo código de função.
post_invocation_app_level Chamado logo após a execução da função ser concluída. O contexto de função, os argumentos de invocação de função e o objeto de retorno de invocação são passados para a extensão. Essa implementação é um bom local para validar se a execução dos ganchos do ciclo de vida foi bem-sucedida.

Extensões de nível de função

Uma extensão que herda de FuncExtensionBase é executada em um gatilho de função específico.

FuncExtensionBase expõe os seguintes métodos de classe abstrata para implementações:

Método Descrição
__init__ O construtor da extensão. Ele é chamado quando uma instância de extensão é inicializada em uma função específica. Ao implementar esse método abstrato, é aconselhável aceitar um parâmetro filename e passá-lo para o método pai super().__init__(filename) para o registro de extensão adequado.
post_function_load Chamado logo após a função ser carregada. O nome da função e o diretório de função são passados para a extensão. Não esqueça que o diretório de funções é somente leitura e toda tentativa de gravar no arquivo local falhará nesse diretório.
pre_invocation Chamado logo antes da função ser disparada. Os argumentos de invocação de função e de contexto de função são passados para a extensão. Geralmente, você pode passar outros atributos no objeto de contexto para serem consumidos pelo código de função.
post_invocation Chamado logo após a execução da função ser concluída. O contexto de função, os argumentos de invocação de função e o objeto de retorno de invocação são passados para a extensão. Essa implementação é um bom local para validar se a execução dos ganchos do ciclo de vida foi bem-sucedida.

Compartilhamento de recursos entre origens

O Azure Functions oferece suporte a compartilhamento de recursos entre origens (CORS). O CORS está configurado no portal e por meio do CLI do Azure. A lista de origens permitidas pelo CORS aplica-se ao nível do aplicativo de funções. Com o CORS habilitado, as respostas incluem o cabeçalho Access-Control-Allow-Origin. Para obter mais informações, consulte Compartilhamento de recursos entre origens.

O CORS (compartilhamento de recursos entre origens) tem suporte completo dos aplicativos de funções do Python.

Assíncrono

Por padrão, uma instância de host para Python pode processar apenas uma invocação de função por vez. Isso ocorre porque o Python é um runtime de thread único. Para um aplicativo de funções que processa um grande número de eventos de E/S ou que está sendo vinculado a E/S, você pode aprimorar significativamente o desempenho executando funções de maneira assíncrona. Para obter mais informações, confira Aprimorar todo o desempenho de aplicativos Python no Azure Functions.

Memória compartilhada (pré-visualização)

Para melhorar a taxa de transferência, o Azure Functions permite que sua função de trabalho de linguagem Python fora de processo compartilhe memória com o processo de host do Functions. Quando seu aplicativo de funções está atingindo gargalos, você pode habilitar a memória compartilhada adicionando uma configuração de aplicativo FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED com um valor de 1. Com a memória compartilhada habilitada, use a configuração DOCKER_SHM_SIZE para definir a memória compartilhada como 268435456, equivalente a 256 MB.

Por exemplo, você pode habilitar a memória compartilhada para reduzir gargalos ao usar suas associações de Armazenamento de Blobs para transferir cargas maiores que 1 MB.

Essa funcionalidade está disponível somente em aplicativos de funções que estão em execução Premium e Dedicada (Serviço de Aplicativo do Azure). Para saber mais, confira Memória compartilhada.

Problemas conhecidos e perguntas frequentes

Confira estes dois guias de solução de problemas comuns:

Confira dois guias de solução de problemas conhecidos com o modelo de programação v2:

Todos os problemas conhecidos e solicitações de recursos são controlados em uma Lista de problemas do GitHub. Se você tiver um problema e não for possível localizá-lo no GitHub, abra um novo problema e inclua uma descrição detalhada dele.

Próximas etapas

Para saber mais, consulte os recursos a seguir:

Problemas com o uso do Python? Conte-nos o que está acontecendo.