Guia do desenvolvedor de Python para o Azure Functions

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.

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.

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.

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) 

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)

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)

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.

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.

Pré-requisitos

• Runtime do Azure Functions versão 4.34 ou posterior.
• Python versão 3.9 ou uma versão com suporte posterior.

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:

• Tipo ContainerClient
• Tipo StorageStreamDownloader

Pré-requisitos

• O runtime do Azure Functions versão 4.34.1 ou posterior.
• Python versão 3.8 ou uma versão com suporte posterior.

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

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

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}".

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)

app = Flask("Test")

@app.route("hello/<name>", methods=["GET"])
def hello(name: str):
    return f"hello {name}"

def main(req: func.HttpRequest, context) -> func.HttpResponse:
  logging.info("Python HTTP trigger function processed a request.")
  return func.WsgiMiddleware(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:

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

# function_app.py

import azure.functions as func 
from flask import Flask, Response 

flask_app = Flask(__name__) 

@flask_app.get("/return_http") 
def return_http(): 
    return Response("<h1>Hello World™</h1>", mimetype="text/html") 

app = func.WsgiFunctionApp(app=flask_app.wsgi_app, 
                           http_auth_level=func.AuthLevel.ANONYMOUS) 

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.

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.

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.

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.