diff --git a/migrations/versions/user_auth_migration.py b/migrations/versions/user_auth_migration.py index 14fbf2c..2612088 100644 --- a/migrations/versions/user_auth_migration.py +++ b/migrations/versions/user_auth_migration.py @@ -7,6 +7,8 @@ Create Date: 2023-10-28 """ from alembic import op import sqlalchemy as sa +from sqlalchemy.engine.reflection import Inspector +from sqlalchemy.exc import OperationalError # revision identifiers, used by Alembic. @@ -16,43 +18,108 @@ branch_labels = None depends_on = None +def table_exists(table_name): + """Check if table exists in the database""" + conn = op.get_bind() + inspector = Inspector.from_engine(conn) + return table_name in inspector.get_table_names() + + +def column_exists(table_name, column_name): + """Check if column exists in the table""" + conn = op.get_bind() + inspector = Inspector.from_engine(conn) + if table_name not in inspector.get_table_names(): + return False + columns = [col['name'] for col in inspector.get_columns(table_name)] + return column_name in columns + + +def index_exists(table_name, index_name): + """Check if index exists in the table""" + conn = op.get_bind() + inspector = Inspector.from_engine(conn) + if table_name not in inspector.get_table_names(): + return False + indexes = [idx['name'] for idx in inspector.get_indexes(table_name)] + return index_name in indexes + + def upgrade(): - # Create users table - op.create_table( - 'users', - sa.Column('id', sa.Integer(), nullable=False), - sa.Column('email', sa.String(), nullable=False), - sa.Column('username', sa.String(), nullable=False), - sa.Column('hashed_password', sa.String(), nullable=False), - sa.Column('is_active', sa.Boolean(), nullable=True, default=True), - sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)')), - sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), - sa.PrimaryKeyConstraint('id') - ) + # Create users table if it doesn't exist + if not table_exists('users'): + op.create_table( + 'users', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('email', sa.String(), nullable=False), + sa.Column('username', sa.String(), nullable=False), + sa.Column('hashed_password', sa.String(), nullable=False), + sa.Column('is_active', sa.Boolean(), nullable=True, default=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)')), + sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint('id') + ) - # Create indexes on users table - op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True) - op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False) - op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True) + # Create indexes on users table if they don't exist + try: + if not index_exists('users', 'ix_users_email'): + op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True) + except (OperationalError, sa.exc.InternalError): + pass - # Add owner_id column to todos table - op.add_column('todos', sa.Column('owner_id', sa.Integer(), nullable=True)) + try: + if not index_exists('users', 'ix_users_id'): + op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False) + except (OperationalError, sa.exc.InternalError): + pass - # Create foreign key constraint - op.create_foreign_key('fk_todos_owner_id_users', 'todos', 'users', ['owner_id'], ['id']) + try: + if not index_exists('users', 'ix_users_username'): + op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True) + except (OperationalError, sa.exc.InternalError): + pass + + # Add owner_id column to todos table if it doesn't exist + if not column_exists('todos', 'owner_id'): + op.add_column('todos', sa.Column('owner_id', sa.Integer(), nullable=True)) + + # Create foreign key constraint if it doesn't exist + try: + op.create_foreign_key('fk_todos_owner_id_users', 'todos', 'users', ['owner_id'], ['id']) + except (OperationalError, sa.exc.InternalError): + pass def downgrade(): - # Drop foreign key constraint - op.drop_constraint('fk_todos_owner_id_users', 'todos', type_='foreignkey') + # Try to drop foreign key constraint + try: + op.drop_constraint('fk_todos_owner_id_users', 'todos', type_='foreignkey') + except (OperationalError, sa.exc.InternalError): + pass - # Drop owner_id column from todos table - op.drop_column('todos', 'owner_id') + # Try to drop owner_id column from todos table + if column_exists('todos', 'owner_id'): + op.drop_column('todos', 'owner_id') - # Drop indexes on users table - op.drop_index(op.f('ix_users_username'), table_name='users') - op.drop_index(op.f('ix_users_id'), table_name='users') - op.drop_index(op.f('ix_users_email'), table_name='users') + # Try to drop indexes on users table + try: + if index_exists('users', 'ix_users_username'): + op.drop_index(op.f('ix_users_username'), table_name='users') + except (OperationalError, sa.exc.InternalError): + pass - # Drop users table - op.drop_table('users') \ No newline at end of file + try: + if index_exists('users', 'ix_users_id'): + op.drop_index(op.f('ix_users_id'), table_name='users') + except (OperationalError, sa.exc.InternalError): + pass + + try: + if index_exists('users', 'ix_users_email'): + op.drop_index(op.f('ix_users_email'), table_name='users') + except (OperationalError, sa.exc.InternalError): + pass + + # Try to drop users table + if table_exists('users'): + op.drop_table('users') \ No newline at end of file