Hızlı Başlangıç: Python Dayanıklı İşlevler uygulaması oluşturma

Python'da durum bilgisi olan sunucusuz iş akışları yazmak için Azure İşlevleri özelliği olan Dayanıklı İşlevler kullanın. Bu hızlı başlangıçta, iki yaygın düzenleme desenini gösteren bir örnek uygulamayı kopyalayıp çalıştıracaksınız:

  • İşlev zinciri: Etkinlikleri sırayla çağırır (Tokyo → Seattle → Londra).
  • Fan-out/fan-in: Beş şehirde etkinlikleri paralel olarak çağırır ve ardından sonuçları toplar.

Sonunda, Durable Task Scheduler öykünücüsünde yerel olarak çalışan her iki orkestrasyona da sahip olacak ve bunların durumunu gösterge panelinde görüntüleyebileceksiniz.

  • Hello Cities örnek projesini kopyalayın ve hazırlayın.
  • Yerel geliştirme için Durable Task Scheduler öykünücüsünü ve Azurite'yi kurup yapılandırın.
  • İşlev uygulamasını çalıştırın ve her iki orkestrasyonu da tetikleyin.
  • Dayanıklı Görev Zamanlayıcı panosunda düzenleme durumunu ve çıktıyı gözden geçirin.

Prerequisites

Dayanıklı Görev Zamanlayıcı öykünücüsünü ayarlayın

Durable Task Scheduler emülatörü, Azure aboneliği olmadan orkestrasyonları test edebilmeniz için yerel bir geliştirme ortamı sağlar. İşlevler konağı ayrıca yerel depolama için Azurite gerektirir.

Her iki konteyneri de başlatın:

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

Tip

Öykünücü çalıştırıldıktan sonra, düzenlemeleri izlemek için adresindeki http://localhost:8082 Dayanıklı Görev Zamanlayıcı panosuna erişebilirsiniz.

Hızlı başlangıç örneğini çalıştırma

  1. Hello Cities örnek dizinine gidin:

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

  2. Sanal ortam oluşturma ve bağımlılıkları yükleme:

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

  3. local.settings.json Dosyanın aşağıdaki yapılandırmayı içerdiğini doğrulayın:

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

  4. İşlev uygulamasını başlatın:

    func start

  5. Ayrı bir terminalde işlev zincirleme düzenlemesini tetikleyin :

    $response = Invoke-RestMethod -Method POST -Uri http://localhost:7071/api/StartChaining
$response

    Yanıt, düzenleme örneğinin durum URL'lerini içerir. statusQueryGetUri Değeri kopyalayın ve çalıştırarak sonucu denetleyin:

    Invoke-RestMethod -Uri $response.statusQueryGetUri

  6. fan-out/fan-in orkestrasyonunu tetikleme:

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

Beklenen çıkış

POST isteği durum URL'lerine sahip bir JSON yanıtı döndürür. Örneğin:

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

statusQueryGetUri sorguladığınızda ve orkestrasyonun runtimeStatus değeri Completed olduğunda, selamlama sonuçlarını output alanında bulabilirsiniz. Zincirleme düzenlemesi şunu döndürür:

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

Fan-out/fan-in orkestrasyonu şunu döndürür:

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

Tip

runtimeStatus, Running veya Pending gösteriyorsa bir an bekleyin ve statusQueryGetUri’ü yeniden sorgulayın.

Orkestrasyon durumunu ve yürütme geçmişini görüntülemek için http://localhost:8082 adresindeki Dayanıklı Görev Zamanlayıcı panosunu açın.

Kodu anlama

Örnek, tüm işlevlerin tek bir dosyada (function_app.py) tanımlandığı Python v2 programlama modelini dekoratörlerle birlikte kullanır.

Aktivite işlevi

say_hello etkinliği bir şehir adını girdi olarak alır ve bir selamlama döndürür:

@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}!"

Orchestrator işlevleri

Zincirleme orkestratör, say_hello öğesini üç şehir için sırayla çağırır:

@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]

fan-out/fan-in orchestrator etkinlikleri paralel olarak planlar:

@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

İstemci işlevleri

HTTP ile tetiklenen istemci işlevleri her düzenlemeyi başlatır. Örneğin, zincirleme başlatıcısı:

@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

Örnek, depolama altyapısı olarak Durable Task Scheduler öykünücüsünü kullanır. Bu, host.json içinde yapılandırılır:

{
  "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)"
  }
}

Kaynakları temizle

İşiniz bittiğinde öykünücü kapsayıcılarını durdurun:

docker stop dtsemulator azurite && docker rm dtsemulator azurite

Python sanal ortamını devre dışı bırakmak için:

deactivate

Sonraki Adımlar

  • Last updated on