I'm trying to improve CI pipeline to prevent situations where SQLAlchemy models are added or changed, but no Alembic migration is written or generated by the commit author from hitting the production branch.
alembic --help
doesn't seem to provide any helpful commands for this case, yet it already has all the metadata required (target_metadata
variable) and the database credentials in env.py
to make this happen.
What would be the best practice for implementing this check in CI?
Alembic can view the status of the database and compare against the table metadata in the application, generating the “obvious” migrations based on a comparison. This is achieved using the --autogenerate option to the alembic revision command, which places so-called candidate migrations into our new migrations file.
There's currently no command to delete migrations from your versions directory, so if you want to completely wipe away all trace of your bad migration, you'll need to delete the version file (like 4c009570237e_add_widget_table.py ) manually.
Right click on alembic. ini and choose git>rollback>ok.
Here's a solution that I use. It's a check that I have implemented as a test.
from alembic.autogenerate import compare_metadata
from alembic.command import upgrade
from alembic.runtime.migration import MigrationContext
from alembic.config import Config
from models.base import Base
def test_migrations_sane():
"""
This test ensures that models defined by SQLAlchemy match what alembic migrations think
the database should look like. If these are different, then once we have constructed
the database via Alembic (via running all migrations) alembic will generate a set of changes to
modify the database to match the schema defined by SQLAlchemy models. If these are the same,
the set of changes is going to be empty. Which is exactly what we want to check.
"""
engine = "SQLAlchemy DB Engine instance"
try:
with engine.connect() as connection:
alembic_conf_file = "location of alembic.ini"
alembic_config = Config(alembic_conf_file)
upgrade(alembic_config, "head")
mc = MigrationContext.configure(connection)
diff = compare_metadata(mc, Base.metadata)
assert diff == []
finally:
with engine.connect() as connection:
# Resetting the DB
connection.execute(
"""
DROP SCHEMA public CASCADE;
CREATE SCHEMA public;
GRANT ALL ON SCHEMA public TO postgres;
GRANT ALL ON SCHEMA public TO public;
"""
)
EDIT: I noticed you linked a library that's supposed to do the same thing. I gave it a go but it seems like it assumes that the database that it's running the check against has to have had alembic run against it. My solution works against a blank db.
There is alembic-autogen-check
tool available: https://pypi.org/project/alembic-autogen-check/ Although it requires a database to be created to check upon.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With