Features: job status table with related DB model, migration files, job status tracking.

Added status wrapper to handle job status updates in the DB.

Author: dkaushik94

src/backend/base/langflow/alembic/versions/00e5ef437a72_changed_job_id_in_jobtable_to_be_uuid_.py

@@ -0,0 +1,35 @@
+"""Changed job_id in JobTable to be UUID instead of string
+
+Revision ID: 00e5ef437a72
+Revises: dd3dd875cd97
+Create Date: 2026-01-25 21:32:44.515850
+
+Phase: MIGRATE
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "00e5ef437a72"  # pragma: allowlist secret
+down_revision: str | None = "dd3dd875cd97"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.alter_column("job_id", existing_type=sa.VARCHAR(), type_=sa.Uuid(), existing_nullable=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.alter_column("job_id", existing_type=sa.Uuid(), type_=sa.VARCHAR(), existing_nullable=False)
+
+    # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/dd3dd875cd97_changed_job_id_in_vertex_build_table_to_.py

@@ -0,0 +1,35 @@
+"""Changed job_id in vertex_build table to be UUID instead of str.
+
+Revision ID: dd3dd875cd97
+Revises: f7a564a956a8
+Create Date: 2026-01-25 20:23:02.795268
+
+Phase: MIGRATE
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "dd3dd875cd97"
+down_revision: str | None = "f7a564a956a8"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+        batch_op.alter_column("job_id", existing_type=sa.VARCHAR(), type_=sa.Uuid(), existing_nullable=True)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+        batch_op.alter_column("job_id", existing_type=sa.Uuid(), type_=sa.VARCHAR(), existing_nullable=True)
+
+    # ### end Alembic commands ###

src/backend/base/langflow/alembic/versions/f7a564a956a8_add_job_id_to_vertex_build_create_job_.py

@@ -0,0 +1,62 @@
+"""Add job_id to vertex_build, create job_id status model
+
+Revision ID: f7a564a956a8
+Revises: 182e5471b900
+Create Date: 2026-01-25 19:07:45.358554
+
+Phase: EXPAND
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+import sqlmodel
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "f7a564a956a8"
+down_revision: str | None = "182e5471b900"  # pragma: allowlist secret
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "job",
+        sa.Column("job_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("flow_id", sa.Uuid(), nullable=False),
+        sa.Column(
+            "status",
+            sa.Enum("queued", "in_progress", "completed", "failed", "cancelled", "timed_out", name="job_status_enum"),
+            nullable=False,
+        ),
+        sa.Column("created_timestamp", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("finished_timestamp", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("job_id"),
+    )
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.create_index(batch_op.f("ix_job_flow_id"), ["flow_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_job_job_id"), ["job_id"], unique=False)
+        batch_op.create_index(batch_op.f("ix_job_status"), ["status"], unique=False)
+
+    with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("job_id", sqlmodel.sql.sqltypes.AutoString(), nullable=True))
+        batch_op.create_index(batch_op.f("ix_vertex_build_job_id"), ["job_id"], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("vertex_build", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_vertex_build_job_id"))
+        batch_op.drop_column("job_id")
+
+    with op.batch_alter_table("job", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_job_status"))
+        batch_op.drop_index(batch_op.f("ix_job_job_id"))
+        batch_op.drop_index(batch_op.f("ix_job_flow_id"))
+
+    op.drop_table("job")
+    # ### end Alembic commands ###

src/backend/base/langflow/api/v2/workflow.py

@@ -24,7 +24,7 @@
 import asyncio
 from copy import deepcopy
 from typing import Annotated
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, status
 from fastapi.responses import StreamingResponse
@@ -62,7 +62,7 @@
 from langflow.services.auth.utils import api_key_security
 from langflow.services.database.models.flow.model import FlowRead
 from langflow.services.database.models.user.model import UserRead
-from langflow.services.deps import get_task_service
+from langflow.services.deps import get_job_service, get_task_service
 
 # Configuration constants
 EXECUTION_TIMEOUT = 300  # 5 minutes default timeout for sync execution
@@ -371,10 +371,13 @@ async def execute_sync_workflow(
     terminal_node_ids = graph.get_terminal_nodes()
 
     # Execute graph - component errors are caught and returned in response body
+    job_service = get_job_service()
     try:
-        task_result, execution_session_id = await run_graph_internal(
+        task_result, execution_session_id = await job_service.with_status_updates(
+            job_id=UUID(job_id),
+            flow_id=UUID(flow_id_str),
+            run_graph_func=run_graph_internal,
             graph=graph,
-            flow_id=flow_id_str,
             session_id=session_id,
             inputs=None,
             outputs=terminal_node_ids,
@@ -444,10 +447,21 @@ async def execute_workflow_background(
 
         # Launch background task
         task_service = get_task_service()
+        job_service = get_job_service()
+
+        # Create job synchronously to ensure it exists before background task starts
+        # and so we can return a valid job status immediately
+        await job_service.create_job(
+            job_id=job_id,
+            flow_id=flow_id_str,
+        )
+
         await task_service.fire_and_forget_task(
-            run_graph_internal,
+            job_service.with_status_updates,
+            job_id=UUID(job_id),
+            flow_id=UUID(flow_id_str),
+            run_graph_func=run_graph_internal,
             graph=graph,
-            flow_id=flow_id_str,
             session_id=session_id,
             inputs=None,
             outputs=terminal_node_ids,
@@ -473,35 +487,86 @@ async def execute_workflow_background(
 )
 async def get_workflow_status(
     api_key_user: Annotated[UserRead, Depends(api_key_security)],  # noqa: ARG001
-    job_id: Annotated[str, Query(description="Job ID to query")],  # noqa: ARG001
-) -> WorkflowJobResponse:
-    """Get workflow job status and results by job ID.
+    flow_id: Annotated[UUID | None, Query(description="Flow ID to filter jobs")] = None,
+    job_id: Annotated[UUID | None, Query(description="Job ID to query")] = None,
+    page: Annotated[int, Query(description="Page number (1-indexed)", ge=1)] = 1,
+    page_size: Annotated[int, Query(description="Number of results per page", ge=1, le=100)] = 10,
+) -> list[WorkflowJobResponse]:
+    """Get workflow job status and results.
 
-    This endpoint allows clients to poll for the status of background workflow executions.
+    This endpoint allows clients to query job status either by:
+    - **flow_id**: Get all jobs for a specific flow (with pagination)
+    - **job_id**: Get a specific job by ID
+
+    At least one of flow_id or job_id must be provided.
 
     Args:
         api_key_user: Authenticated user from API key
-        job_id: The job ID returned from a background execution request
+        flow_id: Optional flow ID to filter jobs
+        job_id: Optional job ID to query specific job
+        page: Page number for pagination (default: 1)
+        page_size: Number of results per page (default: 10, max: 100)
 
     Returns:
-        - WorkflowExecutionResponse: If job is complete or failed
-        - StreamingResponse: If job is still running (for streaming mode)
+        List of WorkflowJobResponse objects
 
     Raises:
         HTTPException:
+            - 400: Neither flow_id nor job_id provided
             - 403: Developer API disabled or unauthorized
-            - 404: Job ID not found
-            - 501: Not yet implemented
-
-    Note:
-        This endpoint is not yet implemented. It will be added in a future release
-        to support background and streaming execution modes.
+            - 404: Job or flow not found
+            - 500: Internal server error
     """
-    # TODO: Implement job status tracking and retrieval
-    # - Store job metadata in database or cache
-    # - Track execution progress and status
-    # - Return appropriate response based on job state
-    raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Workflow status not yet available.")
+    # Validate that at least one filter is provided
+    if not flow_id and not job_id:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail={
+                "error": "Missing required parameter",
+                "code": "MISSING_PARAMETER",
+                "message": "Either flow_id or job_id must be provided",
+            },
+        )
+
+    try:
+        job_service = get_job_service()
+
+        if job_id:
+            # Query by specific job_id (returns single job)
+            # TODO: Implement get_job_by_id in CRUD and service
+            raise HTTPException(
+                status_code=status.HTTP_501_NOT_IMPLEMENTED,
+                detail={
+                    "error": "Not implemented",
+                    "code": "NOT_IMPLEMENTED",
+                    "message": "Query by job_id not yet implemented. Use flow_id instead.",
+                },
+            )
+
+        # Query by flow_id
+        jobs = await job_service.get_jobs_by_flow_id(flow_id=flow_id, page=page, page_size=page_size)
+
+        # Convert to response format
+        return [
+            WorkflowJobResponse(
+                job_id=str(job.job_id),
+                flow_id=str(job.flow_id),
+                status=job.status,
+            )
+            for job in jobs
+        ]
+
+    except HTTPException:
+        raise
+    except Exception as exc:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={
+                "error": "Internal server error",
+                "code": "INTERNAL_SERVER_ERROR",
+                "message": f"Failed to retrieve job status: {exc!s}",
+            },
+        ) from exc
 
 
 @router.post(

src/backend/base/langflow/services/database/models/__init__.py

@@ -2,6 +2,7 @@
 from .file import File
 from .flow import Flow
 from .folder import Folder
+from .jobs import Job
 from .message import MessageTable
 from .transactions import TransactionTable
 from .user import User
@@ -12,6 +13,7 @@
     "File",
     "Flow",
     "Folder",
+    "Job",
     "MessageTable",
     "TransactionTable",
     "User",

src/backend/base/langflow/services/database/models/jobs/__init__.py

@@ -0,0 +1,3 @@
+from .model import Job
+
+__all__ = ["Job"]

src/backend/base/langflow/services/database/models/jobs/crud.py

@@ -0,0 +1,65 @@
+from uuid import UUID
+
+from sqlmodel import select
+from sqlmodel.ext.asyncio.session import AsyncSession
+
+from langflow.services.database.models.jobs.model import Job
+
+
+async def get_jobs_by_flow_id(db: AsyncSession, flow_id: UUID, page: int = 1, size: int = 10) -> list[Job]:
+    """Get jobs by flow ID with pagination.
+
+    Args:
+        db: Async database session
+        flow_id: Flow ID to filter by
+        page: Page number (1-indexed)
+        size: Number of results per page
+
+    Returns:
+        List of Job objects
+    """
+    # Handle UUID comparison for different databases:
+    # - SQLite: stores UUIDs as strings without hyphens
+    # - PostgreSQL: has native UUID type
+    # We need to detect the database type and adjust the query accordingly
+
+    # # Get the database dialect name
+    # dialect_name = db.bind.dialect.name if db.bind else "sqlite"
+
+    # if dialect_name == "sqlite":
+    #     # SQLite stores UUIDs as unhyphenated strings
+    #     from sqlalchemy import cast, String
+
+    #     statement = (
+    #         select(Job)
+    #         .where(cast(Job.flow_id, String) == flow_id.hex)
+    #         .offset((page - 1) * size)
+    #         .limit(size)
+    #     )
+    # else:
+    #     # PostgreSQL and other databases with native UUID support
+    statement = select(Job).where(Job.flow_id == flow_id).offset((page - 1) * size).limit(size)
+
+    result = await db.exec(statement)
+    return list(result.all())
+
+
+async def update_job_status(db: AsyncSession, job_id: str, status: str) -> Job | None:
+    """Update job status.
+
+    Args:
+        db: Async database session
+        job_id: Job ID to update
+        status: New status value
+
+    Returns:
+        Updated Job object or None if not found
+    """
+    result = await db.exec(select(Job).where(Job.job_id == job_id))
+    job = result.first()
+    if job:
+        job.status = status
+        db.add(job)
+        await db.commit()
+        await db.refresh(job)
+    return job