Início Rápido: Criar um aplicativo de Python Durable Functions

Use Durable Functions, um recurso do Azure Functions, para criar fluxos de trabalho sem servidor com estado em Python. Neste início rápido, você clona e executa um aplicativo de exemplo que demonstra dois padrões comuns de orquestração:

Encadeamento de funções: chama as atividades sequencialmente (Tóquio → Seattle → Londres).
Fan-out/fan-in: chama atividades em paralelo em cinco cidades e agrega os resultados.

No final, você terá as duas orquestrações em execução localmente com o emulador do Agendador de Tarefas Duráveis e poderá exibir seu status no painel.

Clone e prepare o projeto de exemplo Hello Cities.
Configure o emulador do Agendador de Tarefas Duráveis e o Azurite para desenvolvimento local.
Execute o aplicativo de função e acione ambas as orquestrações.
Verifique o status e a saída da orquestração no dashboard do Durable Task Scheduler.

Pré-requisitos

Python 3.9+ instalado.
Azure Functions Core Tools v4 ou posterior.
Docker para executar o emulador e o Azurite.
Clone o repositório Durable Task Scheduler GitHub para usar o exemplo de início rápido.

Configurar o emulador do Agendador de Tarefas Duráveis

O emulador Durable Task Scheduler fornece um ambiente de desenvolvimento local para que você possa testar orquestrações sem uma assinatura Azure. O host do Functions também requer o Azurite para armazenamento local.

Inicie ambos os contêineres:

docker run -d --name dtsemulator -p 8080:8080 -p 8082:8082 \
  mcr.microsoft.com/dts/dts-emulator:latest

docker run -d --name azurite -p 10000:10000 -p 10001:10001 -p 10002:10002 \
  mcr.microsoft.com/azure-storage/azurite

Dica

Depois que o emulador estiver em execução, você poderá acessar o painel http://localhost:8082 do Agendador de Tarefas Duráveis para monitorar orquestrações.

Executar o exemplo de início rápido

Navegue até o diretório de exemplo Hello Cities:

cd samples/durable-functions/python/hello-cities

Crie um ambiente virtual e instale as dependências:

python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt

Verifique se o local.settings.json arquivo contém a seguinte configuração:

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "DURABLE_TASK_SCHEDULER_CONNECTION_STRING": "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"
  }
}

Inicie o aplicativo de funções:
```
func start
```
Em um terminal separado, dispare a orquestração de encadeamento de funções :
```
$response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartChaining
$response
```
A resposta contém URLs de status para a instância de orquestração. Copie o valor statusQueryGetUri e execute-o para verificar o resultado:
```
Invoke-RestMethod -Uri $response.statusQueryGetUri
```

Acione a orquestração fan-out/fan-in:

$response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartFanOutFanIn
Invoke-RestMethod -Uri $response.statusQueryGetUri

Resultado esperado

A solicitação POST retorna uma resposta JSON com URLs de status. Por exemplo:

{
  "id": "<instanceId>",
  "statusQueryGetUri": "http://localhost:7071/runtime/webhooks/durabletask/instances/<instanceId>?code=...",
  "sendEventPostUri": "...",
  "terminatePostUri": "...",
  "purgeHistoryDeleteUri": "..."
}

Quando você consulta statusQueryGetUri e o runtimeStatus da orquestração é Completed, você pode encontrar os resultados da saudação no campo output. A orquestração de encadeamento retorna:

{
  "name": "chaining_orchestration",
  "runtimeStatus": "Completed",
  "output": ["Hello Tokyo!", "Hello Seattle!", "Hello London!"]
}

A orquestração fan-out/fan-in retorna:

{
  "name": "fan_out_fan_in_orchestration",
  "runtimeStatus": "Completed",
  "output": ["Hello Tokyo!", "Hello Seattle!", "Hello London!", "Hello Paris!", "Hello Berlin!"]
}

Dica

Se runtimeStatus mostrar Running ou Pending, aguarde um momento e consulte statusQueryGetUri novamente.

Abra o painel do Agendador de Tarefas Duráveis em http://localhost:8082 para ver o status da orquestração e o histórico de execução.

Entender o código

O exemplo usa o modelo de programação Python v2 com decoradores, em que todas as funções são definidas em um único arquivo (function_app.py).

Função de atividade

A say_hello atividade usa um nome de cidade e retorna uma saudação:

@app.activity_trigger(input_name="city")
def say_hello(city: str) -> str:
    """Activity function that returns a greeting for a city."""
    logging.info(f"Saying hello to {city}.")
    return f"Hello {city}!"

Funções de orquestrador

O orquestrador de encadeamento chama say_hello sequencialmente para três cidades:

@app.orchestration_trigger(context_name="context")
def chaining_orchestration(context: df.DurableOrchestrationContext):
    """Function chaining orchestration: calls activities sequentially."""
    result1 = yield context.call_activity("say_hello", "Tokyo")
    result2 = yield context.call_activity("say_hello", "Seattle")
    result3 = yield context.call_activity("say_hello", "London")
    return [result1, result2, result3]

O orquestrador fan-out/fan-in programa as atividades em paralelo:

@app.orchestration_trigger(context_name="context")
def fan_out_fan_in_orchestration(context: df.DurableOrchestrationContext):
    """Fan-out/Fan-in orchestration: calls activities in parallel."""
    cities = ["Tokyo", "Seattle", "London", "Paris", "Berlin"]

    # Fan-out: schedule all activities in parallel
    parallel_tasks = []
    for city in cities:
        task = context.call_activity("say_hello", city)
        parallel_tasks.append(task)

    # Fan-in: wait for all to complete
    results = yield context.task_all(parallel_tasks)
    return results

Funções do cliente

As funções de cliente disparadas por HTTP iniciam cada orquestração. Por exemplo, o iniciador do encadeamento:

@app.route(route="StartChaining", methods=["POST"])
@app.durable_client_input(client_name="client")
async def start_chaining(req: func.HttpRequest, client) -> func.HttpResponse:
    """HTTP trigger to start the function chaining orchestration."""
    instance_id = await client.start_new("chaining_orchestration")
    logging.info(f"Started chaining orchestration with ID = '{instance_id}'.")
    return client.create_check_status_response(req, instance_id)

Configuration

O exemplo usa o emulador do Agendador de Tarefas Duráveis como seu back-end de armazenamento. Isso está configurado em host.json:

{
  "version": "2.0",
  "logging": {
    "logLevel": {
      "DurableTask.Core": "Warning"
    }
  },
  "extensions": {
    "durableTask": {
      "hubName": "default",
      "storageProvider": {
        "type": "azureManaged",
        "connectionStringName": "DURABLE_TASK_SCHEDULER_CONNECTION_STRING"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[4.*, 5.0.0)"
  }
}

Limpar os recursos

Interrompa os contêineres do emulador quando terminar:

docker stop dtsemulator azurite && docker rm dtsemulator azurite

Para desativar o ambiente virtual Python:

deactivate

Próximas Etapas

Saiba mais sobre os padrões comuns de aplicativos do Durable Functions.
Saiba mais sobre os provedores de armazenamento do Durable Functions.

Comentários

Esta página foi útil?

Last updated on 2026-06-05