mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 05:16:40 +08:00
feat(jobs): add durable result and error columns to job table
This commit is contained in:
@ -0,0 +1,44 @@
|
||||
"""add result and error columns to job table.
|
||||
|
||||
Revision ID: 185482a2d715
|
||||
Revises: b7c4d8e9f012
|
||||
Create Date: 2026-06-03 10:00:00.000000
|
||||
|
||||
Phase: EXPAND
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
from sqlalchemy.dialects import postgresql
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision: str = "185482a2d715" # pragma: allowlist secret
|
||||
down_revision: str | None = "b7c4d8e9f012" # pragma: allowlist secret
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
_JSON = sa.JSON().with_variant(postgresql.JSONB(), "postgresql")
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
existing_columns = {col["name"] for col in sa.inspect(conn).get_columns("job")}
|
||||
|
||||
with op.batch_alter_table("job", schema=None) as batch_op:
|
||||
if "result" not in existing_columns:
|
||||
batch_op.add_column(sa.Column("result", _JSON, nullable=True))
|
||||
if "error" not in existing_columns:
|
||||
batch_op.add_column(sa.Column("error", _JSON, nullable=True))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
existing_columns = {col["name"] for col in sa.inspect(conn).get_columns("job")}
|
||||
|
||||
with op.batch_alter_table("job", schema=None) as batch_op:
|
||||
if "error" in existing_columns:
|
||||
batch_op.drop_column("error")
|
||||
if "result" in existing_columns:
|
||||
batch_op.drop_column("result")
|
||||
@ -85,6 +85,19 @@ class JobBase(SQLModel):
|
||||
sa_column=Column(JsonVariant, nullable=True),
|
||||
)
|
||||
|
||||
# Durable terminal payloads for background workflow runs. result holds
|
||||
# the final output blob on COMPLETED; error holds {type, message, ...}
|
||||
# on FAILED/TIMED_OUT. Both nullable so non-workflow jobs and in-flight
|
||||
# rows stay readable.
|
||||
result: dict[str, Any] | None = Field(
|
||||
default=None,
|
||||
sa_column=Column(JsonVariant, nullable=True),
|
||||
)
|
||||
error: dict[str, Any] | None = Field(
|
||||
default=None,
|
||||
sa_column=Column(JsonVariant, nullable=True),
|
||||
)
|
||||
|
||||
|
||||
class Job(JobBase, table=True): # type: ignore[call-arg]
|
||||
__tablename__ = "job"
|
||||
|
||||
@ -0,0 +1,27 @@
|
||||
"""Structure tests for the background-execution migrations.
|
||||
|
||||
Covers job.result/error columns, the job_events table and execution_signals
|
||||
table. Runs on sqlite and (when configured) postgres.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from alembic import command
|
||||
from sqlalchemy import create_engine, inspect
|
||||
|
||||
from .test_migration_execution import _engine_url, _make_alembic_cfg, db_url # noqa: F401
|
||||
|
||||
|
||||
def test_job_has_result_and_error_columns(db_url): # noqa: F811
|
||||
alembic_cfg = _make_alembic_cfg(db_url)
|
||||
command.upgrade(alembic_cfg, "head")
|
||||
|
||||
engine = create_engine(_engine_url(db_url))
|
||||
try:
|
||||
with engine.connect() as connection:
|
||||
columns = {c["name"] for c in inspect(connection).get_columns("job")}
|
||||
finally:
|
||||
engine.dispose()
|
||||
|
||||
assert "result" in columns
|
||||
assert "error" in columns
|
||||
Reference in New Issue
Block a user