mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-07-14 16:36:07 +00:00
fix: port workflow schema migration to alembic
This commit is contained in:
RockChinQ
@@ -0,0 +1,207 @@
|
||||
"""add workflow tables and bot binding fields
|
||||
|
||||
Revision ID: 0009_add_workflow_tables
|
||||
Revises: 0008_mcp_resource_prefs
|
||||
Create Date: 2026-07-01
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision = '0009_add_workflow_tables'
|
||||
down_revision = '0008_mcp_resource_prefs'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def _table_exists(conn: sa.Connection, table_name: str) -> bool:
|
||||
return table_name in sa.inspect(conn).get_table_names()
|
||||
|
||||
|
||||
def _has_column(conn: sa.Connection, table_name: str, column_name: str) -> bool:
|
||||
if not _table_exists(conn, table_name):
|
||||
return False
|
||||
return column_name in {column['name'] for column in sa.inspect(conn).get_columns(table_name)}
|
||||
|
||||
|
||||
def _has_index_for_columns(conn: sa.Connection, table_name: str, columns: tuple[str, ...]) -> bool:
|
||||
if not _table_exists(conn, table_name):
|
||||
return False
|
||||
for index in sa.inspect(conn).get_indexes(table_name):
|
||||
if tuple(index.get('column_names') or ()) == columns:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _ensure_index(conn: sa.Connection, table_name: str, index_name: str, columns: list[str]) -> None:
|
||||
if _has_index_for_columns(conn, table_name, tuple(columns)):
|
||||
return
|
||||
op.create_index(index_name, table_name, columns)
|
||||
|
||||
|
||||
def _create_workflow_tables(conn: sa.Connection) -> None:
|
||||
if not _table_exists(conn, 'workflows'):
|
||||
op.create_table(
|
||||
'workflows',
|
||||
sa.Column('uuid', sa.String(255), primary_key=True),
|
||||
sa.Column('name', sa.String(255), nullable=False),
|
||||
sa.Column('description', sa.Text(), nullable=True),
|
||||
sa.Column('emoji', sa.String(10), nullable=True),
|
||||
sa.Column('version', sa.Integer(), nullable=False, server_default='1'),
|
||||
sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
|
||||
sa.Column('definition', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
|
||||
sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
|
||||
sa.Column(
|
||||
'extensions_preferences',
|
||||
sa.JSON(),
|
||||
nullable=False,
|
||||
server_default=sa.text(
|
||||
'\'{"enable_all_plugins": true, "enable_all_mcp_servers": true, "plugins": [], "mcp_servers": []}\''
|
||||
),
|
||||
),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
)
|
||||
|
||||
if not _table_exists(conn, 'workflow_versions'):
|
||||
op.create_table(
|
||||
'workflow_versions',
|
||||
sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
|
||||
sa.Column('workflow_uuid', sa.String(255), nullable=False),
|
||||
sa.Column('version', sa.Integer(), nullable=False),
|
||||
sa.Column('definition', sa.JSON(), nullable=False),
|
||||
sa.Column('global_config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
sa.Column('created_by', sa.String(255), nullable=True),
|
||||
sa.UniqueConstraint('workflow_uuid', 'version', name='uq_workflow_version'),
|
||||
)
|
||||
|
||||
if not _table_exists(conn, 'workflow_triggers'):
|
||||
op.create_table(
|
||||
'workflow_triggers',
|
||||
sa.Column('uuid', sa.String(255), primary_key=True),
|
||||
sa.Column('workflow_uuid', sa.String(255), nullable=False),
|
||||
sa.Column('type', sa.String(50), nullable=False),
|
||||
sa.Column('config', sa.JSON(), nullable=False, server_default=sa.text("'{}'")),
|
||||
sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
|
||||
sa.Column('priority', sa.Integer(), nullable=False, server_default='0'),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
)
|
||||
|
||||
if not _table_exists(conn, 'workflow_executions'):
|
||||
op.create_table(
|
||||
'workflow_executions',
|
||||
sa.Column('uuid', sa.String(255), primary_key=True),
|
||||
sa.Column('workflow_uuid', sa.String(255), nullable=False),
|
||||
sa.Column('workflow_version', sa.Integer(), nullable=False),
|
||||
sa.Column('status', sa.String(20), nullable=False),
|
||||
sa.Column('trigger_type', sa.String(50), nullable=True),
|
||||
sa.Column('trigger_data', sa.JSON(), nullable=True),
|
||||
sa.Column('variables', sa.JSON(), nullable=True),
|
||||
sa.Column('start_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('end_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('error', sa.Text(), nullable=True),
|
||||
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
|
||||
)
|
||||
|
||||
if not _table_exists(conn, 'workflow_node_executions'):
|
||||
op.create_table(
|
||||
'workflow_node_executions',
|
||||
sa.Column('id', sa.Integer(), primary_key=True, autoincrement=True),
|
||||
sa.Column('execution_uuid', sa.String(255), nullable=False),
|
||||
sa.Column('node_id', sa.String(100), nullable=False),
|
||||
sa.Column('node_type', sa.String(50), nullable=False),
|
||||
sa.Column('status', sa.String(20), nullable=False),
|
||||
sa.Column('inputs', sa.JSON(), nullable=True),
|
||||
sa.Column('outputs', sa.JSON(), nullable=True),
|
||||
sa.Column('start_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('end_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('error', sa.Text(), nullable=True),
|
||||
sa.Column('retry_count', sa.Integer(), nullable=False, server_default='0'),
|
||||
)
|
||||
|
||||
if not _table_exists(conn, 'workflow_scheduled_jobs'):
|
||||
op.create_table(
|
||||
'workflow_scheduled_jobs',
|
||||
sa.Column('uuid', sa.String(255), primary_key=True),
|
||||
sa.Column('trigger_uuid', sa.String(255), nullable=False),
|
||||
sa.Column('cron_expression', sa.String(100), nullable=True),
|
||||
sa.Column('next_run_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('last_run_time', sa.DateTime(), nullable=True),
|
||||
sa.Column('is_enabled', sa.Boolean(), nullable=False, server_default=sa.true()),
|
||||
)
|
||||
|
||||
_ensure_index(conn, 'workflow_versions', 'ix_workflow_versions_workflow_uuid', ['workflow_uuid'])
|
||||
_ensure_index(conn, 'workflow_triggers', 'ix_workflow_triggers_workflow_uuid', ['workflow_uuid'])
|
||||
_ensure_index(conn, 'workflow_executions', 'ix_workflow_executions_workflow_uuid', ['workflow_uuid'])
|
||||
_ensure_index(
|
||||
conn,
|
||||
'workflow_node_executions',
|
||||
'ix_workflow_node_executions_execution_uuid',
|
||||
['execution_uuid'],
|
||||
)
|
||||
_ensure_index(conn, 'workflow_scheduled_jobs', 'ix_workflow_scheduled_jobs_trigger_uuid', ['trigger_uuid'])
|
||||
|
||||
|
||||
def _add_bot_binding_fields(conn: sa.Connection) -> None:
|
||||
if not _table_exists(conn, 'bots'):
|
||||
return
|
||||
|
||||
if not _has_column(conn, 'bots', 'binding_type'):
|
||||
op.add_column(
|
||||
'bots',
|
||||
sa.Column('binding_type', sa.String(32), nullable=False, server_default='pipeline'),
|
||||
)
|
||||
|
||||
if not _has_column(conn, 'bots', 'binding_uuid'):
|
||||
op.add_column('bots', sa.Column('binding_uuid', sa.String(64), nullable=True))
|
||||
|
||||
conn.execute(
|
||||
sa.text("""
|
||||
UPDATE bots
|
||||
SET binding_uuid = use_pipeline_uuid
|
||||
WHERE use_pipeline_uuid IS NOT NULL
|
||||
AND use_pipeline_uuid != ''
|
||||
AND (binding_uuid IS NULL OR binding_uuid = '')
|
||||
""")
|
||||
)
|
||||
conn.execute(
|
||||
sa.text("""
|
||||
UPDATE bots
|
||||
SET binding_type = 'pipeline'
|
||||
WHERE binding_uuid IS NOT NULL
|
||||
AND binding_uuid != ''
|
||||
AND (binding_type IS NULL OR binding_type = '')
|
||||
""")
|
||||
)
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
_create_workflow_tables(conn)
|
||||
_add_bot_binding_fields(conn)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
|
||||
if _has_column(conn, 'bots', 'binding_uuid'):
|
||||
with op.batch_alter_table('bots') as batch_op:
|
||||
batch_op.drop_column('binding_uuid')
|
||||
if _has_column(conn, 'bots', 'binding_type'):
|
||||
with op.batch_alter_table('bots') as batch_op:
|
||||
batch_op.drop_column('binding_type')
|
||||
|
||||
for table_name in (
|
||||
'workflow_scheduled_jobs',
|
||||
'workflow_node_executions',
|
||||
'workflow_executions',
|
||||
'workflow_triggers',
|
||||
'workflow_versions',
|
||||
'workflows',
|
||||
):
|
||||
if _table_exists(conn, table_name):
|
||||
op.drop_table(table_name)
|
||||
@@ -1,158 +0,0 @@
|
||||
"""Add workflow tables and update bot binding fields"""
|
||||
|
||||
import sqlalchemy
|
||||
from .. import migration
|
||||
|
||||
|
||||
@migration.migration_class(26)
|
||||
class DBMigrateWorkflowTables(migration.DBMigration):
|
||||
"""Add workflow tables and update bot binding fields"""
|
||||
|
||||
async def upgrade(self):
|
||||
# Create workflows table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflows (
|
||||
uuid VARCHAR(255) PRIMARY KEY,
|
||||
name VARCHAR(255) NOT NULL,
|
||||
description TEXT,
|
||||
emoji VARCHAR(10) DEFAULT '🔄',
|
||||
version INTEGER NOT NULL DEFAULT 1,
|
||||
is_enabled BOOLEAN NOT NULL DEFAULT 1,
|
||||
definition JSON NOT NULL DEFAULT '{}',
|
||||
global_config JSON NOT NULL DEFAULT '{}',
|
||||
extensions_preferences JSON NOT NULL DEFAULT '{"enable_all_plugins": true, "enable_all_mcp_servers": true, "plugins": [], "mcp_servers": []}',
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create workflow_versions table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflow_versions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
workflow_uuid VARCHAR(255) NOT NULL,
|
||||
version INTEGER NOT NULL,
|
||||
definition JSON NOT NULL,
|
||||
global_config JSON NOT NULL DEFAULT '{}',
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
created_by VARCHAR(255),
|
||||
UNIQUE(workflow_uuid, version)
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create workflow_triggers table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflow_triggers (
|
||||
uuid VARCHAR(255) PRIMARY KEY,
|
||||
workflow_uuid VARCHAR(255) NOT NULL,
|
||||
type VARCHAR(50) NOT NULL,
|
||||
config JSON NOT NULL DEFAULT '{}',
|
||||
is_enabled BOOLEAN NOT NULL DEFAULT 1,
|
||||
priority INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create workflow_executions table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflow_executions (
|
||||
uuid VARCHAR(255) PRIMARY KEY,
|
||||
workflow_uuid VARCHAR(255) NOT NULL,
|
||||
workflow_version INTEGER NOT NULL,
|
||||
status VARCHAR(20) NOT NULL,
|
||||
trigger_type VARCHAR(50),
|
||||
trigger_data JSON,
|
||||
variables JSON,
|
||||
start_time TIMESTAMP,
|
||||
end_time TIMESTAMP,
|
||||
error TEXT,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create workflow_node_executions table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflow_node_executions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
execution_uuid VARCHAR(255) NOT NULL,
|
||||
node_id VARCHAR(100) NOT NULL,
|
||||
node_type VARCHAR(50) NOT NULL,
|
||||
status VARCHAR(20) NOT NULL,
|
||||
inputs JSON,
|
||||
outputs JSON,
|
||||
start_time TIMESTAMP,
|
||||
end_time TIMESTAMP,
|
||||
error TEXT,
|
||||
retry_count INTEGER NOT NULL DEFAULT 0
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create workflow_scheduled_jobs table
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
CREATE TABLE IF NOT EXISTS workflow_scheduled_jobs (
|
||||
uuid VARCHAR(255) PRIMARY KEY,
|
||||
trigger_uuid VARCHAR(255) NOT NULL,
|
||||
cron_expression VARCHAR(100),
|
||||
next_run_time TIMESTAMP,
|
||||
last_run_time TIMESTAMP,
|
||||
is_enabled BOOLEAN NOT NULL DEFAULT 1
|
||||
)
|
||||
""")
|
||||
)
|
||||
|
||||
# Create indexes
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text('CREATE INDEX IF NOT EXISTS idx_workflow_versions_uuid ON workflow_versions(workflow_uuid)')
|
||||
)
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text('CREATE INDEX IF NOT EXISTS idx_workflow_triggers_uuid ON workflow_triggers(workflow_uuid)')
|
||||
)
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text(
|
||||
'CREATE INDEX IF NOT EXISTS idx_workflow_executions_uuid ON workflow_executions(workflow_uuid)'
|
||||
)
|
||||
)
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text(
|
||||
'CREATE INDEX IF NOT EXISTS idx_workflow_node_executions_uuid ON workflow_node_executions(execution_uuid)'
|
||||
)
|
||||
)
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text(
|
||||
'CREATE INDEX IF NOT EXISTS idx_workflow_scheduled_jobs_trigger ON workflow_scheduled_jobs(trigger_uuid)'
|
||||
)
|
||||
)
|
||||
|
||||
# Update bots table: add binding_type column (default to 'pipeline' for backward compatibility)
|
||||
# Check if column exists first (SQLite doesn't support IF NOT EXISTS for columns)
|
||||
try:
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('SELECT binding_type FROM bots LIMIT 1'))
|
||||
except Exception:
|
||||
# Column doesn't exist, add it
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("ALTER TABLE bots ADD COLUMN binding_type VARCHAR(20) NOT NULL DEFAULT 'pipeline'")
|
||||
)
|
||||
|
||||
async def downgrade(self):
|
||||
# Drop tables in reverse order
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_scheduled_jobs'))
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_node_executions'))
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_executions'))
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_triggers'))
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflow_versions'))
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('DROP TABLE IF EXISTS workflows'))
|
||||
|
||||
# Remove binding_type column from bots (SQLite doesn't support DROP COLUMN directly)
|
||||
# This would need a table recreation in SQLite, so we'll skip it in downgrade
|
||||
@@ -1,49 +0,0 @@
|
||||
"""Add binding_uuid field to bots table and migrate data"""
|
||||
|
||||
import sqlalchemy
|
||||
from .. import migration
|
||||
|
||||
|
||||
@migration.migration_class(27)
|
||||
class DBMigrateBotBindingFields(migration.DBMigration):
|
||||
"""Add binding_uuid field to bots table and migrate existing data"""
|
||||
|
||||
async def upgrade(self):
|
||||
# Add binding_uuid column to bots table
|
||||
# Check if column exists first (SQLite doesn't support IF NOT EXISTS for columns)
|
||||
try:
|
||||
await self.ap.persistence_mgr.execute_async(sqlalchemy.text('SELECT binding_uuid FROM bots LIMIT 1'))
|
||||
except Exception:
|
||||
# Column doesn't exist, add it
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text('ALTER TABLE bots ADD COLUMN binding_uuid VARCHAR(64)')
|
||||
)
|
||||
|
||||
# Migrate existing data: copy use_pipeline_uuid to binding_uuid for records
|
||||
# that have a pipeline bound and binding_uuid is not set yet
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
UPDATE bots
|
||||
SET binding_uuid = use_pipeline_uuid
|
||||
WHERE use_pipeline_uuid IS NOT NULL
|
||||
AND use_pipeline_uuid != ''
|
||||
AND (binding_uuid IS NULL OR binding_uuid = '')
|
||||
""")
|
||||
)
|
||||
|
||||
# Ensure binding_type is 'pipeline' for records that were migrated
|
||||
await self.ap.persistence_mgr.execute_async(
|
||||
sqlalchemy.text("""
|
||||
UPDATE bots
|
||||
SET binding_type = 'pipeline'
|
||||
WHERE binding_uuid IS NOT NULL
|
||||
AND binding_uuid != ''
|
||||
AND (binding_type IS NULL OR binding_type = '')
|
||||
""")
|
||||
)
|
||||
|
||||
async def downgrade(self):
|
||||
# SQLite doesn't support DROP COLUMN directly
|
||||
# This would need a table recreation in SQLite, so we'll skip it in downgrade
|
||||
# The column will remain but won't be used
|
||||
pass
|
||||
Reference in New Issue
Block a user