SQLAlchemy and Alembic are a great combination to keep your database in sync with your application as we saw in the last post. But what do you do when you already created some tables with SQLAlchemy that now contain data we cannot lose? Let us look how we can fix this without much extra work.
This post is part of my journey to learn Python. You can find the other parts of this series here. You find the code for this post in my PythonFriday repository on GitHub.
How SQLAlchemy and Alembic work together
If we use SQLAlchemy and Alembic from the start, all our tables are created with migrations. In this case a new developer can take the code, create an empty database and run the migrations to get the same set of tables as everyone else. If this matches your situation, you do not have a problem we try to solve here.
The problem
If you created tables in the past from the SQLAlchemy metadata, you have no migrations to create those tables. You can start to use Alembic, but it will only work with a pre-existing database containing those tables. New developers cannot recreate the database and need a copy of an existing one. Sooner than later this will be a big problem. Therefore, let us fix this now before you start making big changes with Alembic.
A solution
There are different solutions to this problem, but I like one that shows the clear intention of the project to start with a new concept and makes this transition explicit. We create a migration with all existing tables and a helper, that only creates the tables when they do not exist yet. In this case the new developer can start with running the migrations and get the exactly same structure as the team members who have databases since the start of the project.
The only pre-requisite is that everyone who wants to keep their database is on the newest state. You either upgrade to the same level now or create an empty database when the migrations are in place. That applies to the production database as well. If in doubt, start with the production database as your base level.
Create the helper
Inside the alembic folder we create a alembic_helpers.py file with this content:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
# Code based on https://github.com/talkpython/data-driven-web-apps-with-flask from alembic import op from sqlalchemy import engine_from_config from sqlalchemy.engine import reflection def table_does_not_exist(table, schema=None): config = op.get_context().config engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.') insp = reflection.Inspector.from_engine(engine) return insp.has_table(table, schema) == False def table_has_column(table, column): config = op.get_context().config engine = engine_from_config( config.get_section(config.config_ini_section), prefix='sqlalchemy.') insp = reflection.Inspector.from_engine(engine) has_column = False for col in insp.get_columns(table): if column not in col['name']: continue has_column = True return has_column |
This helper uses the Reflection.Inspector in the SQLAlchemy Engine to check if a table exists or not. We invert the answer to get simpler code inside our migrations since we only want to create the table when it does not exist.
Remove existing migrations
You need to get rid of all the existing migrations in alembic/versions. Otherwise Alembic sees that the database is not on the latest version and it will not work. Removing the migrations is not a problem, then Alembic will create all the tables you currently have as one big migration script.
Delete the alembic_version table if it exists or at least delete the row it may contain. Otherwise the checks of Alembic will also prevent you from creating a new migration script for your tables.
Get an empty database
We need an empty database so that Alembic can create a database migration for our “old” tables. This can be as simple as changing the connection string to a non-existing SQLite file in alembic.ini:
|
1 2 3 4 5 |
# the output encoding used when revision files # are written from script.py.mako # output_encoding = utf-8 sqlalchemy.url = sqlite:///./db/BookstoreEMPTY.sqlite |
For other database systems you may need to create an empty database first and then change the connection string.
Auto-generate the migration
We now can tell Alembic to create a migration for the current state of our application.
|
1 |
alembic revision --autogenerate -m "Initial tables" |
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
INFO [alembic.autogenerate.compare] Detected added table ‘Employee’
INFO [alembic.autogenerate.compare] Detected added table ‘Publisher’
INFO [alembic.autogenerate.compare] Detected added table ‘Book’
INFO [alembic.autogenerate.compare] Detected added table ‘BookAuthor’
INFO [alembic.autogenerate.compare] Detected added table ‘BookDetail’
Generating D:\Python\ORM\alembic\versions\347126fc065d_initial_tables.py … done
This creates us the file 347126fc065d_initial_tables.py with CREATE Table statements for all our tables:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 |
... def upgrade(): # ### commands auto generated by Alembic - please adjust! ### op.create_table('Author', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('FirstName', sa.String(), nullable=False), sa.Column('LastName', sa.String(), nullable=False), sa.PrimaryKeyConstraint('Id') ) op.create_table('Employee', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('LastName', sa.String(), nullable=False), sa.Column('FirstName', sa.String(), nullable=False), sa.Column('BirthDate', sa.String(), nullable=True), sa.PrimaryKeyConstraint('Id') ) op.create_table('Publisher', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('Name', sa.String(), nullable=False), sa.PrimaryKeyConstraint('Id') ) ... # ### end Alembic commands ### def downgrade(): # ### commands auto generated by Alembic - please adjust! ### op.drop_table('BookDetail') op.drop_table('BookAuthor') op.drop_table('Book') op.drop_table('Publisher') op.drop_table('Employee') op.drop_table('Author') # ### end Alembic commands ### |
Use the helper
We need to include the location of our helper in the Python path in all migration files that need them. When this is done, we can use it to only create missing tables:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 |
"""Initial tables Revision ID: 347126fc065d Revises: Create Date: 2021-06-12 17:27:18.833477 """ from alembic import op import sqlalchemy as sa import imp import os alembic_helpers = imp.load_source('alembic_helpers', ( os.getcwd() + '/' + op.get_context().script.dir + '/alembic_helpers.py')) # revision identifiers, used by Alembic. revision = '347126fc065d' down_revision = None branch_labels = None depends_on = None def upgrade(): # ### commands auto generated by Alembic - please adjust! ### if(alembic_helpers.table_does_not_exist('Author')): op.create_table('Author', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('FirstName', sa.String(), nullable=False), sa.Column('LastName', sa.String(), nullable=False), sa.PrimaryKeyConstraint('Id') ) if(alembic_helpers.table_does_not_exist('Employee')): op.create_table('Employee', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('LastName', sa.String(), nullable=False), sa.Column('FirstName', sa.String(), nullable=False), sa.Column('BirthDate', sa.String(), nullable=True), sa.PrimaryKeyConstraint('Id') ) if(alembic_helpers.table_does_not_exist('Publisher')): op.create_table('Publisher', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('Name', sa.String(), nullable=False), sa.PrimaryKeyConstraint('Id') ) if(alembic_helpers.table_does_not_exist('Book')): op.create_table('Book', sa.Column('Id', sa.Integer(), autoincrement=True, nullable=False), sa.Column('Title', sa.String(), nullable=False), sa.Column('ISBN', sa.String(length=13), nullable=False), sa.Column('Pages', sa.Integer(), nullable=True), sa.Column('PublishedBy', sa.Integer(), nullable=False), sa.ForeignKeyConstraint(['PublishedBy'], ['Publisher.Id'], ), sa.PrimaryKeyConstraint('Id') ) if(alembic_helpers.table_does_not_exist('BookAuthor')): op.create_table('BookAuthor', sa.Column('BookId', sa.Integer(), nullable=False), sa.Column('AuthorId', sa.Integer(), nullable=False), sa.ForeignKeyConstraint(['AuthorId'], ['Author.Id'], ), sa.ForeignKeyConstraint(['BookId'], ['Book.Id'], ), sa.PrimaryKeyConstraint('BookId', 'AuthorId') ) if(alembic_helpers.table_does_not_exist('BookDetail')): op.create_table('BookDetail', sa.Column('Id', sa.Integer(), nullable=False), sa.Column('Cover', sa.String(), nullable=True), sa.Column('Description', sa.String(), nullable=True), sa.ForeignKeyConstraint(['Id'], ['Book.Id'], ), sa.PrimaryKeyConstraint('Id') ) # ### end Alembic commands ### def downgrade(): # ### commands auto generated by Alembic - please adjust! ### op.drop_table('BookDetail') op.drop_table('BookAuthor') op.drop_table('Book') op.drop_table('Publisher') op.drop_table('Employee') op.drop_table('Author') # ### end Alembic commands ### |
Back to our database
Revert the changes to alembic.ini to use the connection string that points to your database that contains the tables and your data.
Run the migration
We can now run the migration and Alembic will register the migration but not create the tables:
|
1 |
alembic upgrade head |
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
INFO [alembic.runtime.migration] Running upgrade -> 347126fc065d, Initial tables
|
1 |
alembic current |
INFO [alembic.runtime.migration] Context impl SQLiteImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
347126fc065d (head)
If you run the migration in an empty database, it will create the tables. In either way, you now have a database with a registered migration revision and your tables. From here, everyone in the project can apply future changes to the database with Alembic.
Next
With this solution in place, you can change your application in whatever way you seam fit. SQLAlchemy and Alembic will apply your changes from your Python code to the database. Next week we look how we can create an index and use constraints to enforce data integrity.