backend.im/simpletodoapplication-v28h6u
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix migration to handle existing tables and columns

Browse Source 
- Added helper functions to check if tables, columns, and indexes already exist
- Modified upgrade function to skip operations if objects already exist
- Added try-except blocks to handle potential errors during creation operations
- Updated downgrade function to check existence before dropping objects
- Fixed compatibility with databases where tables are already created
This commit is contained in:
Automated Action 2025-05-16 02:16:15 +00:00
parent 03bb5a7929
commit f1b8b8cbee
1 changed files with 97 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

127
migrations/versions/user_auth_migration.py
View File

@ -7,6 +7,8 @@ Create Date: 2023-10-28
""" """
from alembic import op from alembic import op
import sqlalchemy as sa import sqlalchemy as sa
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy.exc import OperationalError
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
@ -16,43 +18,108 @@ branch_labels = None
depends_on = 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(): def upgrade():
# Create users table # Create users table if it doesn't exist
op.create_table( if not table_exists('users'):
'users', op.create_table(
sa.Column('id', sa.Integer(), nullable=False), 'users',
sa.Column('email', sa.String(), nullable=False), sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(), nullable=False), sa.Column('email', sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False), sa.Column('username', sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=True, default=True), sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)')), sa.Column('is_active', sa.Boolean(), nullable=True, default=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)')),
sa.PrimaryKeyConstraint('id') sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
) sa.PrimaryKeyConstraint('id')
)
# Create indexes on users table # Create indexes on users table if they don't exist
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True) try:
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False) if not index_exists('users', 'ix_users_email'):
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True) 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 try:
op.add_column('todos', sa.Column('owner_id', sa.Integer(), nullable=True)) 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 try:
op.create_foreign_key('fk_todos_owner_id_users', 'todos', 'users', ['owner_id'], ['id']) 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(): def downgrade():
# Drop foreign key constraint # Try to drop foreign key constraint
op.drop_constraint('fk_todos_owner_id_users', 'todos', type_='foreignkey') 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 # Try to drop owner_id column from todos table
op.drop_column('todos', 'owner_id') if column_exists('todos', 'owner_id'):
op.drop_column('todos', 'owner_id')
# Drop indexes on users table # Try to drop indexes on users table
op.drop_index(op.f('ix_users_username'), table_name='users') try:
op.drop_index(op.f('ix_users_id'), table_name='users') if index_exists('users', 'ix_users_username'):
op.drop_index(op.f('ix_users_email'), table_name='users') op.drop_index(op.f('ix_users_username'), table_name='users')
except (OperationalError, sa.exc.InternalError):
pass
# Drop users table try:
op.drop_table('users') 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')