Our current way to create the tables in the database when we run the application works fine until we need to extend an existing table. Then SQLAlchemy will not do that for us, and we must make the change manually. With Alembic we have a solution for that problem that works great with SQLAlchemy. Let us add it to our to-do application.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
Install and configure Alembic
You can find all the necessary steps to install and configure Alembic in the post #86 Database Migrations With Alembic and SQLAlchemy. If you want to follow along, run the commands inside the extended_todo folder.
If you installed Alembic in the past, you should make sure that you get the newest version:
1 |
pip install -U alembic |
Application specific configuration
We can initialise Alembic with this command:
1 |
alembic init alembic |
In alembic.ini (next to our main.py) we need to set the path to the database:
1 |
sqlalchemy.url = sqlite:///./db/todo_api.sqlite |
In the alembic folder, we add this code to the env.py file:
1 2 3 4 5 6 7 8 9 10 11 12 13 |
... import sys import os folder = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.insert(0, folder) from data.entities import Task from data.entitybase import EntityBase target_metadata = EntityBase.metadata ... |
Make sure that you remove the existing line that sets target_metadata to None:
1 |
# target_metadata = None |
We need to decide if we want to delete our existing database or if we want to add code to only create the task table if it does not exist (see here for a tutorial and check the Git repository for the syntax for SQLAlchemy 2.x).
I go with the first option and delete the database so that Alembic can later recreate it with the tasks table.
To create the migration for the now deleted database, we can use this command inside the extended_todo folder:
1 |
alembic revision --autogenerate -m "add Task" |
This creates us a migration file inside extended_todo\alembic\versions\ that will create the tasks table when we migrate upwards and that deletes the table when we migrate down:
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 |
"""add Task Revision ID: 6254e63b6309 Revises: Create Date: 2024-07-12 16:04:49.233155 """ from typing import Sequence, Union from alembic import op import sqlalchemy as sa # revision identifiers, used by Alembic. revision: str = '6254e63b6309' down_revision: Union[str, None] = None branch_labels: Union[str, Sequence[str], None] = None depends_on: Union[str, Sequence[str], None] = None def upgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### op.create_table('tasks', sa.Column('id', sa.Integer(), nullable=False), sa.Column('name', sa.String(), nullable=True), sa.Column('priority', sa.Integer(), nullable=True), sa.Column('due_date', sa.Date(), nullable=True), sa.Column('done', sa.Boolean(), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=True), sa.PrimaryKeyConstraint('id') ) # ### end Alembic commands ### def downgrade() -> None: # ### commands auto generated by Alembic - please adjust! ### op.drop_table('tasks') # ### end Alembic commands ### |
We can run the migration with this command:
1 |
alembic upgrade head |
Remove the create tables in FastAPI
In our data/database.py file we need to remove the call to the create_tables() method and the method itself:
1 2 3 4 5 6 7 8 9 10 11 12 |
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from .__all_models import * from .entitybase import EntityBase def create_session_factory(db_file: str) -> sessionmaker: engine = create_engine( 'sqlite:///' + db_file, connect_args={"check_same_thread": False} ) factory = sessionmaker(autocommit=False, autoflush=False, bind=engine) return factory |
This prevents us from accidentally creating tables outside of our Alembic migrations.
New or changed tables
We can add new tables or change existing ones in the entities.py file as we like. When we create a migration, Alembic checks for the differences between the current database and our entities and makes the change for us. We then can apply the migration and the change is in our database.
Fix the tests
We currently use our startup hook to delete the test database and then recreate it from scratch. With the changes to use Alembic that does no longer work. Instead of just deleting the database, we need to get a new database that has the tables in it.
We can create a copy of the current database and put it into a template file:
1 |
cp .\db\todo_api.sqlite .\db\template_test.sqlite |
In our hook, we can delete the existing test database and copy the template database to use it as the new test database:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
import os import shutil def pytest_sessionstart(session): """ Removes the ./db/test_db.sqlite database at the start of the test run """ db_file = os.path.join( os.path.dirname(__file__), '.', 'db', 'test_db.sqlite') template_file = os.path.join( os.path.dirname(__file__), '.', 'db', 'template_test.sqlite') if os.path.isfile(db_file): os.remove(db_file) shutil.copy2(template_file, db_file) print(f"DB {db_file} replaced with template") |
We can now run our tests and they work again. If we change the table structure, we need to update our template database as well. It is a bit of extra work, but that way we get all the flexibility we need.
Next
With Alembic in place, we can evolve our database should the need arise. As history told us, those changes may be closer than we expect. Next week we take our application into a different direction and add a web interface to our API.
2 thoughts on “Python Friday #235: DB Migrations With Alembic and FastAPI”