opensource
/
dify
mirror of https://github.com/langgenius/dify.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: typing

This commit is contained in:
Yeuoly 2025-10-18 20:14:00 +08:00
parent 5fba41688a
commit c7498c3a11
1 changed files with 3 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
api/tasks/async_workflow_tasks.py
View File

@ -45,21 +45,21 @@ SANDBOX_QUEUE = _sandbox_queue
@shared_task(queue=PROFESSIONAL_QUEUE)
def execute_workflow_professional(task_data_dict: dict) -> dict:
def execute_workflow_professional(task_data_dict: dict[str, Any]) -> dict[str, Any]:
"""Execute workflow for professional tier with highest priority"""
task_data = WorkflowTaskData.model_validate(task_data_dict)
return _execute_workflow_common(task_data).model_dump()
@shared_task(queue=TEAM_QUEUE)
def execute_workflow_team(task_data_dict: dict) -> dict:
def execute_workflow_team(task_data_dict: dict[str, Any]) -> dict[str, Any]:
"""Execute workflow for team tier"""
task_data = WorkflowTaskData.model_validate(task_data_dict)
return _execute_workflow_common(task_data).model_dump()
@shared_task(queue=SANDBOX_QUEUE)
def execute_workflow_sandbox(task_data_dict: dict) -> dict:
def execute_workflow_sandbox(task_data_dict: dict[str, Any]) -> dict[str, Any]:
"""Execute workflow for free tier with lower retry limit"""
task_data = WorkflowTaskData.model_validate(task_data_dict)
return _execute_workflow_common(task_data).model_dump()