By Bernat Sampera 5 min read Follow:

Migrate a sqlite database

When running a database is essential to have a migration strategy to be able to add, remove, change tables and columns in a database. In this guide I'll go over this

First I’ll define the file that contains the migrations. As you can see in the _get_migrations function an array is returned with all the migrations that the system has.

This file creates a table schema_migrations that will keep track of the current migrations applied to be able to know when a new one has to be done.

# migrations.py
"""Simple migration system for SQLite database."""

import sqlite3
from pathlib import Path
from typing import List


class Migration:
    """Represents a single database migration."""

    def __init__(self, version: str, description: str, sql: str):
        self.version = version
        self.description = description
        self.sql = sql


class MigrationManager:
    """Manages database migrations."""

    def __init__(self, db_path: str):
        self.db_path = Path(db_path)
        self.migrations = self._get_migrations()

    def _get_migrations(self) -> List[Migration]:
        """Define all migrations here."""
        return [
            Migration(
                version="001",
                description="Add user_id column to glossary_entries",
                sql="ALTER TABLE glossary_entries ADD COLUMN user_id TEXT DEFAULT NULL;",
            ),
            # Add more migrations here as needed
            # Migration(
            #     version="002", 
            #     description="Example future migration",
            #     sql="ALTER TABLE glossary_entries ADD COLUMN example_column TEXT;",
            # ),
        ]

    def _ensure_migration_table(self, conn: sqlite3.Connection):
        """Create the migration tracking table if it doesn't exist."""
        conn.execute("""
            CREATE TABLE IF NOT EXISTS schema_migrations (
                version TEXT PRIMARY KEY,
                description TEXT NOT NULL,
                applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()

    def _get_applied_migrations(self, conn: sqlite3.Connection) -> set:
        """Get list of already applied migrations."""
        cursor = conn.execute("SELECT version FROM schema_migrations")
        return {row[0] for row in cursor.fetchall()}

    def run_migrations(self) -> List[str]:
        """Run all pending migrations."""
        applied = []

        with sqlite3.connect(self.db_path) as conn:
            self._ensure_migration_table(conn)
            applied_migrations = self._get_applied_migrations(conn)

            for migration in self.migrations:
                if migration.version not in applied_migrations:
                    try:
                        # Execute the migration SQL
                        conn.execute(migration.sql)
                        conn.commit()
                        
                        # Record the migration as applied
                        conn.execute(
                            "INSERT INTO schema_migrations (version, description) VALUES (?, ?)",
                            (migration.version, migration.description),
                        )
                        conn.commit()
                        applied.append(f"{migration.version}: {migration.description}")
                        
                    except sqlite3.OperationalError as e:
                        # Handle cases where column already exists or other conflicts
                        if "duplicate column name" in str(e).lower():
                            # Column already exists, just record the migration as applied
                            conn.execute(
                                "INSERT INTO schema_migrations (version, description) VALUES (?, ?)",
                                (migration.version, migration.description),
                            )
                            conn.commit()
                            applied.append(f"{migration.version}: {migration.description} (already applied)")
                        else:
                            raise e

        return applied

    def get_migration_status(self) -> dict:
        """Get status of all migrations."""
        with sqlite3.connect(self.db_path) as conn:
            self._ensure_migration_table(conn)
            applied_migrations = self._get_applied_migrations(conn)

            status = {}
            for migration in self.migrations:
                status[migration.version] = {
                    "description": migration.description,
                    "applied": migration.version in applied_migrations,
                }

            return status

To be able to run this we’ll add a call to this migrations manager every time the database is initialised.

    def _init_database(self):
        """Initialize the SQLite database with all required tables."""
        with self._get_connection() as conn:
            cursor = conn.cursor()

            # Create all tables
            cursor.execute(GLOSSARY_TABLE_SCHEMA)
            cursor.execute(GLOSSARY_INDEX_SCHEMA)
            cursor.execute(USER_IP_TABLE_SCHEMA)
            cursor.execute(WAITLIST_TABLE_SCHEMA)

            # Add more table creation statements here as needed
            # cursor.execute(ANALYTICS_TABLE_SCHEMA)

            conn.commit()

        # Run migrations after initial table creation
        self._run_migrations()

    def _run_migrations(self):
        """Run database migrations."""
        migration_manager = MigrationManager(str(self.db_path))
        applied_migrations = migration_manager.run_migrations()

        if applied_migrations:
            print(f"Applied {len(applied_migrations)} migrations:")
            for migration in applied_migrations:
                print(f"  - {migration}")
        else:
            print("No pending migrations found.")

Let's connect !!

Get in touch if you want updates, examples, and insights on how AI agents, Langchain and more are evolving and where they’re going next.