Custom and simple postgresql migration script in Python

Thursday, October 09, 2025

For one of my side projects I'm using Flask with flask-sqlalchemy to handle database connections.

During development, I needed to run a migration but I found that Flask-Migrate was overkill for this side project, so I looked for a simpler alternative.

Create a migrations folder containing two files:

- applied.json
- migrations_config.py

The applied.json file will serve as the source of truth for the applied migrations. This file should be excluded from your version control system.

The migrations_config.py file will contain the available migrations. For example:

MIGRATIONS = [
    {
        "name": "2025_09_26_add_notes_fields",
        "description": "Add notes field to Video and Channel tables",
        "sql": [
            "ALTER TABLE videos ADD COLUMN notes TEXT",
            "ALTER TABLE channels ADD COLUMN notes TEXT",
        ],
    },
]

Create the migrations.py file in your project root. Please make sure to modify the import paths according to your project:

from app.server import flask_app, db
from app.migrations.migrations_config import MIGRATIONS
import json
import os
from sqlalchemy import text


def run_migrations():
    applied_file = "app/migrations/applied.json"

    # Create migrations directory if it doesn't exist
    os.makedirs(os.path.dirname(applied_file), exist_ok=True)

    # Load applied migrations
    if os.path.exists(applied_file):
        with open(applied_file, "r") as f:
            applied = set(json.load(f))
    else:
        applied = set()

    print("Starting database migrations...")
    print("-" * 50)

    with flask_app.app_context():
        for migration in MIGRATIONS:
            name = migration["name"]

            if name in applied:
                print(f"Skipping {name} (already applied)")
                continue

            print(f"Running {name}: {migration['description']}")

            try:
                for sql in migration["sql"]:
                    db.session.execute(text(sql))
                    print(f"   ✓ {sql}")

                # Commit the changes
                db.session.commit()

                applied.add(name)
                print(f"✅ {name} completed")

            except Exception as e:
                print(f"❌ {name} failed: {e}")
                # Rollback on error
                db.session.rollback()
                break

        # Create directory and save applied migrations
        os.makedirs(os.path.dirname(applied_file), exist_ok=True)
        with open(applied_file, "w") as f:
            json.dump(list(applied), f, indent=2)

    print("-" * 50)
    print("Migration process completed!")


if __name__ == "__main__":
    run_migrations()

The final flow to create a migration would be:

  1. Modify the migrations_config.py to include the new migration in the array.
  2. Run python migrations.py

It's not perfect, but it's enough for my use case :)

Happy coding!