feat: redesign the visual pipelines module as a validated DAG (atomized nodes, typed contracts)

DashAI/alembic/versions/0dc6fd4637a3_merge_dag_pipelines_and_develop_heads.py

@@ -0,0 +1,26 @@
+"""merge dag-pipelines and develop heads
+
+Revision ID: 0dc6fd4637a3
+Revises: b4f9e70098e7, c4f7a8b2d1e3
+Create Date: 2026-04-02 19:32:55.182155
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '0dc6fd4637a3'
+down_revision: Union[str, None] = ('b4f9e70098e7', 'c4f7a8b2d1e3')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

DashAI/alembic/versions/50e54749a557_merge_develop_and_dag_pipelines_heads.py

@@ -0,0 +1,23 @@
+"""merge develop and dag-pipelines heads
+
+Revision ID: 50e54749a557
+Revises: a5f2c71e9d40, b1cf2532fbbf
+Create Date: 2026-09-21 14:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+# revision identifiers, used by Alembic.
+revision: str = "50e54749a557"
+down_revision: Union[str, None] = ("a5f2c71e9d40", "b1cf2532fbbf")
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

DashAI/alembic/versions/66a67b766b49_merge_develop_and_dag_pipelines_heads.py

@@ -0,0 +1,23 @@
+"""merge develop and dag-pipelines heads
+
+Revision ID: 66a67b766b49
+Revises: 50e54749a557, f4a91c62d8e7
+Create Date: 2026-09-29 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+# revision identifiers, used by Alembic.
+revision: str = "66a67b766b49"
+down_revision: Union[str, None] = ("50e54749a557", "f4a91c62d8e7")
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

DashAI/alembic/versions/b1cf2532fbbf_merge_pipeline_tracking_and_folder_.py

@@ -0,0 +1,25 @@
+"""merge pipeline tracking and folder system heads
+
+Revision ID: b1cf2532fbbf
+Revises: d9c8f2a1b7e4, f1a2b3c4d5e6
+Create Date: 2026-06-25 00:13:45.379060
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = 'b1cf2532fbbf'
+down_revision: Union[str, None] = ('d9c8f2a1b7e4', 'f1a2b3c4d5e6')
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

DashAI/alembic/versions/c4f7a8b2d1e3_atomize_train_node_into_split_task_metrics.py

@@ -0,0 +1,107 @@
+"""Atomize train node into split/task/metrics tables.
+
+Revision ID: c4f7a8b2d1e3
+Revises: 98791b9df4f0
+Create Date: 2026-03-25 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "c4f7a8b2d1e3"
+down_revision: Union[str, None] = "98791b9df4f0"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def _table_exists(table_name: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def _column_exists(table_name: str, column_name: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return any(
+        column["name"] == column_name
+        for column in inspector.get_columns(table_name)
+    )
+
+
+def upgrade() -> None:
+    if _table_exists("pipeline"):
+        if not _column_exists("pipeline", "split_data"):
+            op.add_column("pipeline", sa.Column("split_data", sa.JSON(), nullable=True))
+        if not _column_exists("pipeline", "task_and_model"):
+            op.add_column(
+                "pipeline", sa.Column("task_and_model", sa.JSON(), nullable=True)
+            )
+        if not _column_exists("pipeline", "metrics_result"):
+            op.add_column(
+                "pipeline", sa.Column("metrics_result", sa.JSON(), nullable=True)
+            )
+
+    if not _table_exists("split_data_node"):
+        op.create_table(
+            "split_data_node",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("pipeline_id", sa.Integer(), nullable=False),
+            sa.Column("input_columns", sa.JSON(), nullable=False),
+            sa.Column("output_columns", sa.JSON(), nullable=False),
+            sa.Column("splits", sa.JSON(), nullable=False),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.Column("last_modified", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(["pipeline_id"], ["pipeline.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    if not _table_exists("task_and_model_node"):
+        op.create_table(
+            "task_and_model_node",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("pipeline_id", sa.Integer(), nullable=False),
+            sa.Column("task_name", sa.String(), nullable=False),
+            sa.Column("model_name", sa.String(), nullable=False),
+            sa.Column("parameters", sa.JSON(), nullable=True),
+            sa.Column("model_path", sa.String(), nullable=True),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.Column("last_modified", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(["pipeline_id"], ["pipeline.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    if not _table_exists("metrics_node"):
+        op.create_table(
+            "metrics_node",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("pipeline_id", sa.Integer(), nullable=False),
+            sa.Column("metric_names", sa.JSON(), nullable=False),
+            sa.Column("results", sa.JSON(), nullable=True),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.Column("last_modified", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(["pipeline_id"], ["pipeline.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+
+def downgrade() -> None:
+    if _table_exists("metrics_node"):
+        op.drop_table("metrics_node")
+
+    if _table_exists("task_and_model_node"):
+        op.drop_table("task_and_model_node")
+
+    if _table_exists("split_data_node"):
+        op.drop_table("split_data_node")
+
+    if _table_exists("pipeline"):
+        with op.batch_alter_table("pipeline", schema=None) as batch_op:
+            if _column_exists("pipeline", "metrics_result"):
+                batch_op.drop_column("metrics_result")
+            if _column_exists("pipeline", "task_and_model"):
+                batch_op.drop_column("task_and_model")
+            if _column_exists("pipeline", "split_data"):
+                batch_op.drop_column("split_data")

DashAI/alembic/versions/d9c8f2a1b7e4_add_pipeline_run_tracking.py

@@ -0,0 +1,108 @@
+"""Add pipeline run tracking tables.
+
+Revision ID: d9c8f2a1b7e4
+Revises: 0dc6fd4637a3
+Create Date: 2026-04-10 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "d9c8f2a1b7e4"
+down_revision: Union[str, None] = "0dc6fd4637a3"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+def _table_exists(table_name: str) -> bool:
+    inspector = sa.inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+def upgrade() -> None:
+    if not _table_exists("pipeline_run"):
+        op.create_table(
+            "pipeline_run",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("pipeline_id", sa.Integer(), nullable=False),
+            sa.Column("steps", sa.JSON(), nullable=True),
+            sa.Column("edges", sa.JSON(), nullable=True),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.Column("last_modified", sa.DateTime(), nullable=False),
+            sa.Column("delivery_time", sa.DateTime(), nullable=True),
+            sa.Column("start_time", sa.DateTime(), nullable=True),
+            sa.Column("end_time", sa.DateTime(), nullable=True),
+            sa.Column(
+                "status",
+                sa.Enum(
+                    "NOT_STARTED",
+                    "DELIVERED",
+                    "STARTED",
+                    "FINISHED",
+                    "ERROR",
+                    name="pipelinerunstatus",
+                ),
+                nullable=False,
+            ),
+            sa.Column("error_message", sa.String(), nullable=True),
+            sa.ForeignKeyConstraint(["pipeline_id"], ["pipeline.id"], ondelete="CASCADE"),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    if not _table_exists("pipeline_node_run"):
+        op.create_table(
+            "pipeline_node_run",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("pipeline_run_id", sa.Integer(), nullable=False),
+            sa.Column("node_id", sa.String(), nullable=False),
+            sa.Column("node_type", sa.String(), nullable=False),
+            sa.Column("config", sa.JSON(), nullable=True),
+            sa.Column("input", sa.JSON(), nullable=True),
+            sa.Column("output", sa.JSON(), nullable=True),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.Column("last_modified", sa.DateTime(), nullable=False),
+            sa.Column("delivery_time", sa.DateTime(), nullable=True),
+            sa.Column("start_time", sa.DateTime(), nullable=True),
+            sa.Column("end_time", sa.DateTime(), nullable=True),
+            sa.Column(
+                "status",
+                sa.Enum(
+                    "NOT_STARTED",
+                    "DELIVERED",
+                    "STARTED",
+                    "FINISHED",
+                    "ERROR",
+                    name="noderunstatus",
+                ),
+                nullable=False,
+            ),
+            sa.Column("error_message", sa.String(), nullable=True),
+            sa.ForeignKeyConstraint(
+                ["pipeline_run_id"], ["pipeline_run.id"], ondelete="CASCADE"
+            ),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+    if not _table_exists("pipeline_node_artifact"):
+        op.create_table(
+            "pipeline_node_artifact",
+            sa.Column("id", sa.Integer(), nullable=False),
+            sa.Column("node_run_id", sa.Integer(), nullable=False),
+            sa.Column("key", sa.String(), nullable=False),
+            sa.Column("value", sa.JSON(), nullable=True),
+            sa.Column("created", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["node_run_id"], ["pipeline_node_run.id"], ondelete="CASCADE"
+            ),
+            sa.PrimaryKeyConstraint("id"),
+        )
+
+def downgrade() -> None:
+    if _table_exists("pipeline_node_artifact"):
+        op.drop_table("pipeline_node_artifact")
+
+    if _table_exists("pipeline_node_run"):
+        op.drop_table("pipeline_node_run")
+
+    if _table_exists("pipeline_run"):
+        op.drop_table("pipeline_run")