src/backend/base/langflow/worker.py

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from asgiref.sync import async_to_sync
from celery.exceptions import SoftTimeLimitExceeded

from langflow.core.celery_app import celery_app

if TYPE_CHECKING:
    from langflow.graph.vertex.base import Vertex


@celery_app.task(acks_late=True)
def test_celery(word: str) -> str:
    return f"test task return {word}"


@celery_app.task(bind=True, soft_time_limit=30, max_retries=3)
def build_vertex(self, vertex: Vertex) -> Vertex:
    """Build a vertex."""
    try:
        vertex.task_id = self.request.id
        async_to_sync(vertex.build)()
    except SoftTimeLimitExceeded as e:
        raise self.retry(exc=SoftTimeLimitExceeded("Task took too long"), countdown=2) from e
    return vertex


@celery_app.task(acks_late=True)
def process_graph_cached_task() -> dict[str, Any]:
    msg = "This task is not implemented yet"
    raise NotImplementedError(msg)