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.")
Related Posts
Migrating local setup from ollama to llama.cpp
Ollama is a great tool to run local llm models, but it's not the fastest and sometimes has some bugs. To run everything smoother you can use llama.cpp. This guide shows how to use existing ollama models for llama.cpp
Minimal self hosted E2E Setup with Playwright and Allure (Docker)
his minimal stack shows how to do that with Playwright for tests, Allure for reports, Cronicle for scheduling, and optional self‑hosting via Coolify
How to Fix Docker Orphaned Layers and Reclaim Lost Disk Space
Learn how to identify and fix Docker's with Coolify orphaned layer problem that wastes disk space. Step-by-step guide to safely reset Docker data directory and reclaim gigabytes of storage on your VPS server.
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.