I had an already existing database where I had to run migrations.
In attempts to do so, I used alembic and was able to make the migration.
Step 0: Model already exists.
# models.py
USERS = Table('user', admin_meta,
Column('column1', ...),
Column('column2', ...),
Column('column3', ...),
)
STEP 1: Add column, make migration, upgrade
# models.py
USERS = Table('user', admin_meta,
Column('column1', ...),
Column('column2', ...),
Column('column3', ...),
Column('column4', ...), # add new column
)
create custom migrations:
alembic -c database/alembic.ini revision -m "Add backup column"
The migration file:
revision: str = '2aa4282617d4'
...
def upgrade():
op.add_column('users', sa.Column('column4', sa.DateTime))
def downgrade():
op.drop_column('users', 'column4')
Finally, run upgrade.
alembic -c database/alembic.ini upgrade
Step 2: Testing
This is where the problem is. I am testing using an inmemory database and am still loading the models from models.py. So when the database is created in memory, it has the updated models. i.e. model with column4 added.
My idea for testing was simple, start at current/head(because that is what the models have) and go back all the way to step 0, come back all the way to current.
def test_upgrade(dbclient):
# migrate down to the required one first
with dbclient.engine.connect() as connection:
config.attributes['connection'] = connection
alembic.command.downgrade(config, "base")
alembic.command.upgrade(config, "head")
This fails with the error:
FAILED tests/unit/test_migrations.py::test_upgrade - sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) duplicate column name: column4
So in some ways the downgrade didn’t work and it just tried to upgrade from the current state.
What am I doing wrong here? Any leads regarding this would be hugely appreciated.
Thanks in advance.
Long and hard battles were fought before I could figure out that there is a stamp command that will take me to any point in the migration I want to be at. So this works:
config = alembic.config.Config("database/alembic.ini")
config.set_main_option('script_location', 'database/migration')
def test_upgrade(dbclient):
# migrate down to the required one first
with dbclient.engine.connect() as connection:
config.attributes['connection'] = connection
alembic.command.stamp(config, "head")
alembic.command.downgrade(config, 'base')
alembic.command.upgrade(config, "head")