cfbcc8bd79
- تنفيذ مكون ProcedureSelection لتمكين المستخدمين من اختيار الإجراءات من قائمة، وإدارة الاختيارات، ومعالجة الإجراءات المرفوضة. - إنشاء مكون StepProgress لعرض تقدم معالج متعدد الخطوات بشكل مرئي. - تعريف أنواع مشتركة للإجراءات، وخطوات التدفق، ورسائل الدردشة في ملف types.ts. - إضافة اختبارات وحدة لخطافات useFileUpload و useTaskPolling لضمان الأداء السليم ومعالجة الأخطاء. - تنفيذ اختبارات واجهة برمجة التطبيقات (API) للتحقق من تنسيقات نقاط النهاية وضمان اتساق ربط الواجهة الأمامية بالخلفية.
42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
"""Task status polling endpoint."""
|
|
from flask import Blueprint, jsonify
|
|
from celery.result import AsyncResult
|
|
|
|
from app.extensions import celery
|
|
from app.middleware.rate_limiter import limiter
|
|
|
|
tasks_bp = Blueprint("tasks", __name__)
|
|
|
|
|
|
@tasks_bp.route("/<task_id>/status", methods=["GET"])
|
|
@limiter.limit("300/minute", override_defaults=True)
|
|
def get_task_status(task_id: str):
|
|
"""
|
|
Get the status of an async task.
|
|
|
|
Returns:
|
|
JSON with task state and result (if completed)
|
|
"""
|
|
result = AsyncResult(task_id, app=celery)
|
|
|
|
response = {
|
|
"task_id": task_id,
|
|
"state": result.state,
|
|
}
|
|
|
|
if result.state == "PENDING":
|
|
response["progress"] = "Task is waiting in queue..."
|
|
|
|
elif result.state == "PROCESSING":
|
|
meta = result.info or {}
|
|
response["progress"] = meta.get("step", "Processing...")
|
|
|
|
elif result.state == "SUCCESS":
|
|
task_result = result.result or {}
|
|
response["result"] = task_result
|
|
|
|
elif result.state == "FAILURE":
|
|
response["error"] = str(result.info) if result.info else "Task failed."
|
|
|
|
return jsonify(response)
|