البدء السريع: أنشئ تطبيق Python Durable Functions

استخدم Durable Functions، وهي ميزة في ميزة دالات Azure، لكتابة سير عمل بدون خادم ذو حالة في Python. في هذا البدء السريع، تقوم باستنساخ وتشغيل تطبيق عينات يوضح نمطين شائعين من التوزيع الأوركسترالي:

  • تسلسل الدوال: يستدعي الأنشطة بشكل متسلسل (طوكيو → سياتل → لندن).
  • الخروج من اللاعبين/الداخل: يقوم بأنشطة المكالمات بالتوازي عبر خمس مدن، ثم يجمع النتائج.

بحلول النهاية، سيكون لديك كلا النظامين الموسيقيين يعملان محليا باستخدام محاكي Durable Task Scheduler ويمكنك عرض حالتهما في لوحة التحكم.

  • قم باستنساخ وتحضير مشروع العينات لمشروع Hello Cities.
  • قم بإعداد محاكي Durable Task Scheduler وAzurite للتطوير المحلي.
  • شغل تطبيق الوظائف وفعل كلا التنسيقين.
  • راجع حالة التوزيع والنتائج في لوحة تحكم جدولة المهام الدائمة.

المتطلبات المسبقه

  • Python 3.9+ مثبت.
  • دالات Azure Core Tools الإصدار الرابع أو أحدث.
  • Docker لتشغيل المحاكي وAzurite.
  • قم باستنساخ ><جدولة المهام المتطورة GitHub المستودع لاستخدام عينة البدء السريع.

إعداد محاكي جدولة المهام الدائمة

يوفر محاكي جدولة المهام Durable TasksScheduler بيئة تطوير محلية حتى تتمكن من اختبار التنسيقات الموسيقية دون اشتراك Azure. يحتاج مضيف الوظائف أيضا إلى Azurite للتخزين المحلي.

ابدأ كلا الحاويين:

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

نصيحة

بمجرد تشغيل المحاكي، يمكنك الوصول إلى لوحة تحكم Durable Task Scheduler لمراقبة http://localhost:8082 التنسيقات.

شغل عينة البدء السريع

  1. انتقل إلى دليل نماذج Hello Cities:

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

  2. قم بإنشاء بيئة افتراضية وقم بتثبيت التبعيات:

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

  3. تحقق من أن الملف local.settings.json يحتوي على التكوين التالي:

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

  4. شغل تطبيق الوظائف:

    func start

  5. في محطة منفصلة، قم بتفعيل تنسيق سلسلة الدالة :

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

    تحتوي الاستجابة على روابط الحالة لنسخة التنسيق. انسخ القيمة statusQueryGetUri وشغلها للتحقق من النتيجة:

    Invoke-RestMethod -Uri $response.statusQueryGetUri

  6. تشغيل توزيع التوزيع الموسيقي بين التوسع والدخول إلى المروحة :

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

الإنتاج المتوقع

يعيد طلب POST استجابة JSON مع عناوين الحالة. على سبيل المثال:

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

عندما تستفسر statusQueryGetUri وتكون نتائج التوزيع runtimeStatusCompletedهو ، يمكنك العثور على نتائج التحية في output الحقل. تعود التوزيع الموسيقي المتسلسل:

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

تعود التوزيع الموسيقي بين الجمهور والانضمام:

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

نصيحة

إذا runtimeStatus ظهر Running السؤال أو Pendingانتظر لحظة واسأل statusQueryGetUri مرة أخرى.

افتح لوحة تحكم جدولة المهام الدائمة على ال http://localhost:8082 لعرض حالة التوزيع وسجل التنفيذ.

افهم الشيفرة

تستخدم العينة نموذج البرمجة Python v2 مع الديكورينات، حيث تعرف جميع الدوال في ملف واحد (function_app.py).

دالة النشاط

يأخذ النشاط say_hello اسم مدينة ويعيد التحية:

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

دالات المنسق

يقوم منسق التسلسل بالاتصال say_hello بثلاث مدن:

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

يقوم الموزع الخارج/الداخل بجدول الأنشطة بالتوازي:

@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

دالات العميل

تبدأ وظائف العميل التي تفعل بواسطة HTTP كل عملية تنسيق. على سبيل المثال، بادئ السلسلة:

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

التكوين

تستخدم العينة محاكي Durable Task Scheduler كخلفية للتخزين. يتم تكوين هذا في 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)"
  }
}

تنظيف الموارد

أوقف حاويات المحاكي عندما تنتهي:

docker stop dtsemulator azurite && docker rm dtsemulator azurite

لتعطيل بيئة Python الافتراضية:

deactivate

الخطوات التالية

  • Last updated on