docs/development/adding-new-databases.md

Adding New Databases

This guide explains how to add a new SQLite database to the application with automatic migration support.

Overview

The application uses a centralized migration runner that automatically runs migrations for any database during initialization. This ensures:

• All databases get migrations applied automatically on startup
• No need to duplicate migration runner code
• Consistent migration behavior across all databases
• Easy to add new databases with migration support

Quick Start

To add a new database with automatic migration support:

1. Create your database initialization function that creates the schema
2. Call run_migrations_if_needed() at the end of initialization
3. Done! Migrations will run automatically

Example: Adding a New Database

Let's say you want to add a new analytics.db database:

Step 1: Create Schema File

Create fastapi_app/lib/core/analytics_schema.py:

"""
Analytics database schema.
"""

import sqlite3
from pathlib import Path

CREATE_EVENTS_TABLE = """
CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    event_type TEXT NOT NULL,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    data TEXT
)
"""

def initialize_analytics_db(conn: sqlite3.Connection, logger=None, db_path=None) -> None:
    """
    Initialize analytics database schema.

    Creates tables and runs any pending migrations.

    Args:
        conn: SQLite database connection
        logger: Optional logger instance
        db_path: Optional path to database file (needed for migrations)
    """
    try:
        cursor = conn.cursor()

        if logger:
            logger.info("Creating analytics tables...")

        # Create tables
        cursor.execute(CREATE_EVENTS_TABLE)

        conn.commit()

        if logger:
            logger.info("Analytics database schema initialized")

        # Run migrations if db_path provided
        if db_path:
            from pathlib import Path
            from .migration_runner import run_migrations_if_needed
            from .migrations.versions import ANALYTICS_MIGRATIONS  # Create a new list for this database

            run_migrations_if_needed(
                db_path=Path(db_path),
                migrations=ANALYTICS_MIGRATIONS,
                logger=logger
            )

    except sqlite3.Error as e:
        if logger:
            logger.error(f"Failed to initialize analytics database: {e}")
        raise

Step 2: Create Database Manager (Optional)

If you want a database manager class (recommended for complex databases):

"""
Analytics database manager.
"""

import sqlite3
import queue
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from .analytics_schema import initialize_analytics_db
from . import sqlite_utils


class AnalyticsDB:
    """
    Manages analytics database connections with pooling.
    
    Implements connection pooling and safe WAL mode initialization
    similar to the main DatabaseManager.
    """

    def __init__(self, db_path: Path, logger=None):
        self.db_path = db_path
        self.logger = logger
        self._pool = queue.Queue()
        self._ensure_db_exists()

    def _ensure_db_exists(self) -> None:
        """Ensure database and schema exist with migrations."""
        self.db_path.parent.mkdir(parents=True, exist_ok=True)

        # Use per-database lock to prevent concurrent schema initialization
        with sqlite_utils.with_db_lock(self.db_path):
            # Use raw connection for initialization to set WAL mode explicitly
            conn = sqlite3.connect(str(self.db_path), timeout=60.0, isolation_level=None)
            try:
                conn.execute("PRAGMA journal_mode = WAL")
                conn.execute("PRAGMA foreign_keys = ON")
                initialize_analytics_db(conn, self.logger, db_path=self.db_path)
            finally:
                conn.close()

    @contextmanager
    def get_connection(self) -> Generator[sqlite3.Connection, None, None]:
        """
        Context manager for database connections with pooling.
        """
        try:
            conn = self._pool.get(block=False)
        except queue.Empty:
            conn = sqlite3.connect(
                str(self.db_path), 
                timeout=60.0, 
                check_same_thread=False, 
                isolation_level=None
            )
            conn.row_factory = sqlite3.Row
            conn.execute("PRAGMA foreign_keys = ON")

        try:
            yield conn
        finally:
            # Rollback any uncommitted changes to ensure clean state for next use
            try:
                conn.rollback()
            except sqlite3.OperationalError:
                pass
            self._pool.put(conn)

Step 3: Initialize in Application Startup

In fastapi_app/main.py, add initialization in the lifespan function:

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application startup and shutdown lifecycle"""
    # ... existing initialization code ...

    # Initialize analytics database
    from .lib.analytics_db import AnalyticsDB
    analytics_db_path = settings.db_dir / "analytics.db"
    try:
        analytics_db = AnalyticsDB(analytics_db_path, logger)
        logger.info(f"Analytics database initialized: {analytics_db_path}")
    except Exception as e:
        logger.error(f"Error initializing analytics database: {e}")
        raise

    yield

That's it! Migrations will now run automatically for your new database on application startup.

How It Works

The run_migrations_if_needed() function:

1. Checks if migrations are needed by examining the migration_history table
2. Compares registered migrations against what's been applied
3. Runs pending migrations if any are found
4. Caches the result per-process to avoid re-checking on subsequent calls
5. Returns the number of migrations applied (0 if none needed)

Migration System Integration

Migrations are organized by target database in fastapi_app/lib/migrations/versions/__init__.py:

• Database-specific lists: LOCKS_MIGRATIONS, METADATA_MIGRATIONS, etc.
• Global list: ALL_MIGRATIONS (for tools that need the complete list)
• Each database uses its specific migration list to avoid unnecessary checks

When adding a new database, create a new migration list for it in fastapi_app/lib/migrations/versions/__init__.py:

ANALYTICS_MIGRATIONS = [
    Migration003AnalyticsIndexes,
    Migration005AnalyticsCleanup,
]

Example Migration That Targets Specific Database

class Migration003AnalyticsIndexes(Migration):
    """Add indexes to analytics database."""

    @property
    def version(self) -> int:
        return 3

    @property
    def description(self) -> str:
        return "Add indexes to analytics.events table"

    def check_can_apply(self, conn: sqlite3.Connection) -> bool:
        """Only apply if events table exists."""
        cursor = conn.execute("""
            SELECT name FROM sqlite_master
            WHERE type='table' AND name='events'
        """)
        if not cursor.fetchone():
            self.logger.info("Events table does not exist, skipping migration")
            return False
        return True

    def upgrade(self, conn: sqlite3.Connection) -> None:
        """Add indexes to events table."""
        conn.execute("CREATE INDEX IF NOT EXISTS idx_event_type ON events(event_type)")
        conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON events(timestamp)")

Testing Your Database

Always test your database initialization:

def test_analytics_db_initialization():
    import tempfile
    import shutil
    from pathlib import Path
    from fastapi_app.lib.repository.analytics_db import AnalyticsDB
    from fastapi_app.lib.core.migration_runner import reset_migration_cache
    import logging

    logger = logging.getLogger('test')
    temp_dir = Path(tempfile.mkdtemp())

    try:
        # Test initialization
        db_path = temp_dir / 'test_analytics.db'
        db = AnalyticsDB(db_path, logger)

        # Verify schema
        import sqlite3
        with sqlite3.connect(str(db_path)) as conn:
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            )
            tables = [row[0] for row in cursor.fetchall()]
            assert 'events' in tables
            assert 'migration_history' in tables

        print("✓ Database initialized correctly")
    finally:
        shutil.rmtree(temp_dir)

Choosing a Journal Mode

SQLite supports different journal modes. Choose based on your database's characteristics:

WAL Mode (Default)

Use for databases with high concurrency and frequent reads:

conn.execute("PRAGMA journal_mode = WAL")

Use when: High read concurrency, frequent queries, larger databases.

DELETE Mode

Use for simple databases with infrequent writes:

conn.execute("PRAGMA journal_mode = DELETE")

Use when:

• Small databases with infrequent writes
• Short-lived data (like locks or temporary state)
• Databases that don't benefit from WAL's read concurrency
• When rapid concurrent access during tests causes WAL file corruption

Example: The locks.db database uses DELETE mode because it's small, has infrequent writes, and WAL mode caused "disk I/O error" issues during rapid test execution. See fastapi_app/lib/core/locking.py for implementation.

Always Set Busy Timeout

Regardless of journal mode, always set a busy timeout to prevent immediate failures:

conn.execute("PRAGMA busy_timeout = 30000")  # 30 seconds

Best Practices

1. Always pass db_path to your initialization function - This enables migrations
2. Call run_migrations_if_needed() at the end of schema initialization - This ensures tables exist before migrations run
3. Use check_can_apply() in migrations to target specific databases
4. Test with a fresh database to ensure initialization works correctly
5. Document your schema in the schema file
6. Choose the right journal mode - Use WAL for high-concurrency, DELETE for simple low-write databases
7. Always set busy_timeout - Prevents "database is locked" errors

Reference

• Migration system: docs/development/migrations.md
• Migration runner: fastapi_app/lib/core/migration_runner.py
• Database connections guide: docs/code-assistant/database-connections.md
• Example WAL database: fastapi_app/lib/core/database.py (metadata.db)
• Example DELETE database: fastapi_app/lib/core/locking.py (locks.db)