WIP: feat(api): draft var cleanup task

api/tasks/workflow_draft_var_tasks.py

@@ -0,0 +1,27 @@
+"""
+Celery tasks for asynchronous workflow execution storage operations.
+
+These tasks provide asynchronous storage capabilities for workflow execution data,
+improving performance by offloading storage operations to background workers.
+"""
+
+import logging
+
+from celery import shared_task  # type: ignore[import-untyped]
+from sqlalchemy.orm import Session
+
+from extensions.ext_database import db
+
+_logger = logging.getLogger(__name__)
+
+from services.workflow_draft_variable_service import DraftVarFileDeletion, WorkflowDraftVariableService
+
+
+@shared_task(queue="workflow_draft_var", bind=True, max_retries=3, default_retry_delay=60)
+def save_workflow_execution_task(
+    self,
+    deletions: list[DraftVarFileDeletion],
+):
+    with Session(bind=db.engine) as session, session.begin():
+        srv = WorkflowDraftVariableService(session=session)
+        srv.delete_workflow_draft_variable_file(deletions=deletions)