"""202512151837 Revision ID: 64ddbf3c3bcc Revises: 2a46f1e9a590 Create Date: 2025-12-15 18:37:42.649720 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa from sqlalchemy import inspect from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. revision: str = '64ddbf3c3bcc' down_revision: Union[str, None] = '2a46f1e9a590' branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def table_exists(table_name: str) -> bool: """检查表是否存在""" bind = op.get_bind() inspector = inspect(bind) return table_name in inspector.get_table_names() def column_exists(table_name: str, column_name: str) -> bool: """检查列是否存在""" bind = op.get_bind() inspector = inspect(bind) if not table_exists(table_name): return False columns = [col['name'] for col in inspector.get_columns(table_name)] return column_name in columns def index_exists(table_name: str, index_name: str) -> bool: """检查索引是否存在""" bind = op.get_bind() inspector = inspect(bind) if not table_exists(table_name): return False indexes = [idx['name'] for idx in inspector.get_indexes(table_name)] return index_name in indexes def constraint_exists(table_name: str, constraint_name: str) -> bool: """检查约束是否存在(外键、唯一约束等)""" bind = op.get_bind() inspector = inspect(bind) if not table_exists(table_name): return False # 检查外键约束 foreign_keys = [fk['name'] for fk in inspector.get_foreign_keys(table_name) if fk['name']] if constraint_name in foreign_keys: return True # 检查唯一约束 unique_constraints = [uc['name'] for uc in inspector.get_unique_constraints(table_name) if uc['name']] if constraint_name in unique_constraints: return True # 检查检查约束 check_constraints = [cc['name'] for cc in inspector.get_check_constraints(table_name) if cc['name']] if constraint_name in check_constraints: return True return False def trigger_exists(trigger_name: str) -> bool: """检查触发器是否存在(PostgreSQL)""" bind = op.get_bind() result = bind.execute(sa.text( "SELECT EXISTS (SELECT 1 FROM pg_trigger WHERE tgname = :trigger_name)" ), {"trigger_name": trigger_name}) return result.scalar() def sequence_exists(sequence_name: str) -> bool: """检查序列是否存在(PostgreSQL)""" bind = op.get_bind() result = bind.execute(sa.text( "SELECT EXISTS (SELECT 1 FROM pg_class WHERE relkind = 'S' AND relname = :sequence_name)" ), {"sequence_name": sequence_name}) return result.scalar() def enum_exists(enum_name: str) -> bool: """检查枚举类型是否存在(PostgreSQL)""" bind = op.get_bind() result = bind.execute(sa.text( "SELECT EXISTS (SELECT 1 FROM pg_type WHERE typname = :enum_name)" ), {"enum_name": enum_name}) return result.scalar() def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### op.create_table('workflow_configs', sa.Column('id', sa.UUID(), nullable=False), sa.Column('app_id', sa.UUID(), nullable=False), sa.Column('nodes', postgresql.JSONB(astext_type=sa.Text()), nullable=False), sa.Column('edges', postgresql.JSONB(astext_type=sa.Text()), nullable=False), sa.Column('variables', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('execution_config', postgresql.JSONB(astext_type=sa.Text()), nullable=False), sa.Column('triggers', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('is_active', sa.Boolean(), nullable=False), sa.Column('created_at', sa.DateTime(), nullable=False), sa.Column('updated_at', sa.DateTime(), nullable=False), sa.ForeignKeyConstraint(['app_id'], ['apps.id'], ondelete='CASCADE'), sa.PrimaryKeyConstraint('id') ) with op.batch_alter_table('workflow_configs', schema=None) as batch_op: batch_op.create_index(batch_op.f('ix_workflow_configs_app_id'), ['app_id'], unique=True) batch_op.create_index(batch_op.f('ix_workflow_configs_id'), ['id'], unique=False) op.create_table('workflow_executions', sa.Column('id', sa.UUID(), nullable=False), sa.Column('workflow_config_id', sa.UUID(), nullable=False), sa.Column('app_id', sa.UUID(), nullable=False), sa.Column('conversation_id', sa.UUID(), nullable=True), sa.Column('execution_id', sa.String(length=100), nullable=False), sa.Column('trigger_type', sa.String(length=20), nullable=False), sa.Column('triggered_by', sa.UUID(), nullable=True), sa.Column('input_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('output_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('context', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('status', sa.String(length=20), nullable=False), sa.Column('error_message', sa.Text(), nullable=True), sa.Column('error_node_id', sa.String(length=100), nullable=True), sa.Column('started_at', sa.DateTime(), nullable=False), sa.Column('completed_at', sa.DateTime(), nullable=True), sa.Column('elapsed_time', sa.Float(), nullable=True), sa.Column('token_usage', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('meta_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=False), sa.ForeignKeyConstraint(['app_id'], ['apps.id'], ondelete='CASCADE'), sa.ForeignKeyConstraint(['conversation_id'], ['conversations.id'], ondelete='SET NULL'), sa.ForeignKeyConstraint(['triggered_by'], ['users.id'], ), sa.ForeignKeyConstraint(['workflow_config_id'], ['workflow_configs.id'], ondelete='CASCADE'), sa.PrimaryKeyConstraint('id') ) with op.batch_alter_table('workflow_executions', schema=None) as batch_op: batch_op.create_index(batch_op.f('ix_workflow_executions_app_id'), ['app_id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_executions_conversation_id'), ['conversation_id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_executions_execution_id'), ['execution_id'], unique=True) batch_op.create_index(batch_op.f('ix_workflow_executions_id'), ['id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_executions_started_at'), ['started_at'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_executions_status'), ['status'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_executions_workflow_config_id'), ['workflow_config_id'], unique=False) op.create_table('workflow_node_executions', sa.Column('id', sa.UUID(), nullable=False), sa.Column('execution_id', sa.UUID(), nullable=False), sa.Column('node_id', sa.String(length=100), nullable=False), sa.Column('node_type', sa.String(length=20), nullable=False), sa.Column('node_name', sa.String(length=100), nullable=True), sa.Column('execution_order', sa.Integer(), nullable=False), sa.Column('retry_count', sa.Integer(), nullable=False), sa.Column('input_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('output_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('status', sa.String(length=20), nullable=False), sa.Column('error_message', sa.Text(), nullable=True), sa.Column('started_at', sa.DateTime(), nullable=False), sa.Column('completed_at', sa.DateTime(), nullable=True), sa.Column('elapsed_time', sa.Float(), nullable=True), sa.Column('token_usage', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('cache_hit', sa.Boolean(), nullable=True), sa.Column('cache_key', sa.String(length=255), nullable=True), sa.Column('meta_data', postgresql.JSONB(astext_type=sa.Text()), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=False), sa.ForeignKeyConstraint(['execution_id'], ['workflow_executions.id'], ondelete='CASCADE'), sa.PrimaryKeyConstraint('id') ) with op.batch_alter_table('workflow_node_executions', schema=None) as batch_op: batch_op.create_index(batch_op.f('ix_workflow_node_executions_execution_id'), ['execution_id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_node_executions_id'), ['id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_node_executions_node_id'), ['node_id'], unique=False) batch_op.create_index(batch_op.f('ix_workflow_node_executions_status'), ['status'], unique=False) # ### end Alembic commands ### def downgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### with op.batch_alter_table('workflow_node_executions', schema=None) as batch_op: batch_op.drop_index(batch_op.f('ix_workflow_node_executions_status')) batch_op.drop_index(batch_op.f('ix_workflow_node_executions_node_id')) batch_op.drop_index(batch_op.f('ix_workflow_node_executions_id')) batch_op.drop_index(batch_op.f('ix_workflow_node_executions_execution_id')) op.drop_table('workflow_node_executions') with op.batch_alter_table('workflow_executions', schema=None) as batch_op: batch_op.drop_index(batch_op.f('ix_workflow_executions_workflow_config_id')) batch_op.drop_index(batch_op.f('ix_workflow_executions_status')) batch_op.drop_index(batch_op.f('ix_workflow_executions_started_at')) batch_op.drop_index(batch_op.f('ix_workflow_executions_id')) batch_op.drop_index(batch_op.f('ix_workflow_executions_execution_id')) batch_op.drop_index(batch_op.f('ix_workflow_executions_conversation_id')) batch_op.drop_index(batch_op.f('ix_workflow_executions_app_id')) op.drop_table('workflow_executions') with op.batch_alter_table('workflow_configs', schema=None) as batch_op: batch_op.drop_index(batch_op.f('ix_workflow_configs_id')) batch_op.drop_index(batch_op.f('ix_workflow_configs_app_id')) op.drop_table('workflow_configs') # ### end Alembic commands ###