---
title: "HeliosDB Nano SQLite Migration Patterns"
---

# HeliosDB Nano SQLite Migration Patterns

Real-world examples for migrating common Python applications and frameworks to HeliosDB Nano.

---

## Table of Contents

1. [Django ORM Migration](#django-orm-migration)
2. [SQLAlchemy Migration](#sqlalchemy-migration)
3. [Raw SQL Application Migration](#raw-sql-application-migration)
4. [Flask Application Migration](#flask-application-migration)
5. [FastAPI Migration](#fastapi-migration)
6. [Asyncio/Concurrent Application Migration](#asyncioconcurrent-application-migration)
7. [Jupyter Notebook Transition](#jupyter-notebook-transition)
8. [Production Deployment Patterns](#production-deployment-patterns)

---

## Django ORM Migration

Django applications can use HeliosDB Nano with minimal configuration changes.

### Step 1: Install a PostgreSQL driver

There is no dedicated `heliosdb-django` package. Run `heliosdb-nano start` and use
Django's built-in PostgreSQL backend over the wire protocol:

```bash
pip install psycopg2-binary
```

### Step 2: Update Django Settings

**Before (settings.py):**
```python
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}
```

**After (settings.py):**
```python
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'postgres',
        'USER': 'postgres',
        'PASSWORD': 's3cret',      # as passed to heliosdb-nano start --password
        'HOST': '127.0.0.1',
        'PORT': '5432',
    }
}
```

### Step 3: Run Migrations (Existing Database)

```bash
# Your existing migrations work as-is!
python manage.py migrate

# Verify database integrity
python manage.py check --database default

# Run your tests
python manage.py test
```

### Step 4: Leverage Advanced Features (Optional)

There is no Django field type for vectors; add the column and index with a
`RunSQL` migration and query it with raw SQL:

```python
# migrations/0002_vector.py
from django.db import migrations

class Migration(migrations.Migration):
    dependencies = [("shop", "0001_initial")]
    operations = [
        migrations.RunSQL(
            "ALTER TABLE shop_product ADD COLUMN description_embedding VECTOR(384);"
            "CREATE INDEX vec_idx ON shop_product USING hnsw (description_embedding vector_cosine_ops);"
        ),
    ]

# views.py
from django.shortcuts import render
from .models import Product

def search_products(request):
    query_embedding = generate_embedding(request.GET.get("q", ""))
    vec = "[" + ",".join(map(str, query_embedding)) + "]"
    similar_products = Product.objects.raw(
        "SELECT * FROM shop_product "
        "ORDER BY description_embedding <=> %s::vector LIMIT 10",
        [vec],
    )
    return render(request, "products.html", {"products": similar_products})
```

### Step 5: Testing Django + HeliosDB Nano

```python
# tests.py
from django.test import TestCase
from .models import Product

class ProductTestCase(TestCase):
    def test_concurrent_writes(self):
        """Test that concurrent writes don't cause locks"""
        import threading

        def create_product(i):
            Product.objects.create(
                name=f"Product {i}",
                description=f"Description {i}"
            )

        # Create 10 products concurrently
        threads = [threading.Thread(target=create_product, args=(i,)) for i in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        # All 10 should succeed (no locks!)
        self.assertEqual(Product.objects.count(), 10)

    def test_time_travel_query(self):
        """Test historical queries"""
        from django.utils import timezone
        from datetime import timedelta

        # Create product
        product = Product.objects.create(name="Widget", description="Original")

        # Update product
        product.description = "Updated"
        product.save()

        # Query historical state (HeliosDB Nano specific) with AS OF
        past_time = timezone.now() - timedelta(seconds=5)
        historical_product = list(Product.objects.raw(
            "SELECT * FROM shop_product AS OF TIMESTAMP %s WHERE id = %s",
            [past_time.isoformat(), product.id],
        ))[0]

        self.assertEqual(historical_product.description, "Original")
```

---

## SQLAlchemy Migration

SQLAlchemy works with HeliosDB Nano's SQLite compatibility layer.

### Step 1: Update Connection String

**Before:**
```python
from sqlalchemy import create_engine

engine = create_engine('sqlite:///myapp.db')
```

**After:**
```python
from sqlalchemy import create_engine

# Option 1: Use HeliosDB Nano dialect
engine = create_engine('heliosdb:///myapp.db')

# Option 2: Keep sqlite:// with custom creator
import heliosdb_sqlite
engine = create_engine(
    'sqlite:///myapp.db',
    creator=lambda: heliosdb_sqlite.connect('myapp.db')
)
```

### Step 2: Full Application Example

**Before (SQLite):**
```python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# SQLite engine
engine = create_engine('sqlite:///users.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Create user
new_user = User(name='Alice', email='alice@example.com')
session.add(new_user)
session.commit()

# Query users
users = session.query(User).all()
for user in users:
    print(f"{user.name}: {user.email}")
```

**After (HeliosDB Nano):**
```python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

# HeliosDB Nano engine (only change!)
engine = create_engine('heliosdb:///users.db')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Everything else is identical
new_user = User(name='Alice', email='alice@example.com')
session.add(new_user)
session.commit()

users = session.query(User).all()
for user in users:
    print(f"{user.name}: {user.email}")
```

### Step 3: Advanced Features with SQLAlchemy

```python
from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from heliosdb_sqlalchemy import VectorType, similarity_search

Base = declarative_base()

class Document(Base):
    __tablename__ = 'documents'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(Text)
    # Add vector embedding column
    embedding = Column(VectorType(384))

engine = create_engine('heliosdb:///docs.db', echo=True)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

# Insert document with embedding
doc = Document(
    title="Machine Learning Guide",
    content="Introduction to neural networks...",
    embedding=[0.1, 0.2, 0.3, ...]  # 384-dimensional vector
)
session.add(doc)
session.commit()

# Semantic search
query_embedding = [0.15, 0.25, 0.35, ...]
similar_docs = similarity_search(
    session,
    Document,
    Document.embedding,
    query_embedding,
    limit=5
)

for doc in similar_docs:
    print(f"Similar: {doc.title}")
```

### Step 4: Connection Pooling for Production

```python
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# Production configuration with connection pooling
engine = create_engine(
    'heliosdb:///production.db',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,  # Verify connections before use
    echo=False,  # Disable SQL logging in production
)

# Use context manager for sessions
from contextlib import contextmanager

@contextmanager
def get_session():
    session = Session()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Usage
with get_session() as session:
    user = session.query(User).filter_by(email='alice@example.com').first()
    user.name = 'Alice Smith'
    # Automatically committed on context exit
```

---

## Raw SQL Application Migration

For applications using raw SQL queries (no ORM).

### Before (Raw SQLite):

```python
import sqlite3

class UserDatabase:
    def __init__(self, db_path):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()

    def create_user(self, username, email):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO users (username, email) VALUES (?, ?)',
            (username, email)
        )
        user_id = cursor.lastrowid
        conn.commit()
        conn.close()
        return user_id

    def get_user(self, user_id):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
        user = cursor.fetchone()
        conn.close()
        return user

    def update_user(self, user_id, username=None, email=None):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        if username:
            cursor.execute('UPDATE users SET username = ? WHERE id = ?', (username, user_id))
        if email:
            cursor.execute('UPDATE users SET email = ? WHERE id = ?', (email, user_id))

        conn.commit()
        conn.close()

    def delete_user(self, user_id):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
        conn.commit()
        conn.close()

# Usage
db = UserDatabase('users.db')
user_id = db.create_user('alice', 'alice@example.com')
print(db.get_user(user_id))
```

### After (HeliosDB Nano):

```python
import heliosdb_sqlite as sqlite3  # <- Only change at the top!

class UserDatabase:
    def __init__(self, db_path):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()

    # ... rest of the code is IDENTICAL ...

    def create_user(self, username, email):
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        cursor.execute(
            'INSERT INTO users (username, email) VALUES (?, ?)',
            (username, email)
        )
        user_id = cursor.lastrowid
        conn.commit()
        conn.close()
        return user_id

    # All other methods remain unchanged!

# Usage is identical
db = UserDatabase('users.db')
user_id = db.create_user('alice', 'alice@example.com')
print(db.get_user(user_id))
```

### Improved Version with Connection Reuse:

```python
import heliosdb_sqlite as sqlite3
from contextlib import contextmanager

class UserDatabase:
    def __init__(self, db_path):
        self.db_path = db_path
        self._init_db()

    @contextmanager
    def get_connection(self):
        """Context manager for database connections"""
        conn = sqlite3.connect(self.db_path)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def _init_db(self):
        with self.get_connection() as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT UNIQUE NOT NULL,
                    email TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')

    def create_user(self, username, email):
        with self.get_connection() as conn:
            cursor = conn.execute(
                'INSERT INTO users (username, email) VALUES (?, ?)',
                (username, email)
            )
            return cursor.lastrowid

    def get_user(self, user_id):
        with self.get_connection() as conn:
            cursor = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,))
            return cursor.fetchone()

    def bulk_create_users(self, users):
        """Efficiently insert multiple users (works great with HeliosDB Nano)"""
        with self.get_connection() as conn:
            conn.executemany(
                'INSERT INTO users (username, email) VALUES (?, ?)',
                users
            )
```

---

## Flask Application Migration

### Before (Flask + SQLite):

```python
from flask import Flask, request, jsonify
import sqlite3

app = Flask(__name__)
DATABASE = 'app.db'

def get_db():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

@app.route('/users', methods=['GET'])
def get_users():
    conn = get_db()
    cursor = conn.execute('SELECT * FROM users')
    users = [dict(row) for row in cursor.fetchall()]
    conn.close()
    return jsonify(users)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.json
    conn = get_db()
    cursor = conn.execute(
        'INSERT INTO users (name, email) VALUES (?, ?)',
        (data['name'], data['email'])
    )
    user_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return jsonify({'id': user_id}), 201

if __name__ == '__main__':
    app.run(debug=True)
```

### After (Flask + HeliosDB Nano):

```python
from flask import Flask, request, jsonify, g
import heliosdb_sqlite as sqlite3  # <- Only change!

app = Flask(__name__)
DATABASE = 'app.db'

def get_db():
    """Get database connection (reuse within request)"""
    if 'db' not in g:
        g.db = sqlite3.connect(DATABASE)
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(error):
    """Close database connection at end of request"""
    db = g.pop('db', None)
    if db is not None:
        db.close()

@app.route('/users', methods=['GET'])
def get_users():
    conn = get_db()
    cursor = conn.execute('SELECT * FROM users')
    users = [dict(row) for row in cursor.fetchall()]
    return jsonify(users)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.json
    conn = get_db()
    cursor = conn.execute(
        'INSERT INTO users (name, email) VALUES (?, ?)',
        (data['name'], data['email'])
    )
    user_id = cursor.lastrowid
    conn.commit()
    return jsonify({'id': user_id}), 201

if __name__ == '__main__':
    app.run(debug=True)
```

### Flask + HeliosDB Nano Advanced Features:

```python
from flask import Flask, request, jsonify, g
import heliosdb_sqlite as sqlite3

app = Flask(__name__)
app.config['DATABASE'] = 'app.db'
app.config['HELIOSDB_MODE'] = 'hybrid'  # or 'server'

def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect(
            app.config['DATABASE'],
            heliosdb_mode=app.config.get('HELIOSDB_MODE', 'advanced')
        )
        g.db.row_factory = sqlite3.Row
    return g.db

@app.route('/search', methods=['GET'])
def semantic_search():
    """Semantic search using vector similarity"""
    query = request.args.get('q', '')

    # Generate embedding for query
    query_embedding = generate_embedding(query)

    conn = get_db()
    # Use HeliosDB Nano vector search
    cursor = conn.execute('''
        SELECT id, title, content,
               vector_distance(embedding, ?) as distance
        FROM articles
        ORDER BY distance
        LIMIT 10
    ''', (query_embedding,))

    results = [dict(row) for row in cursor.fetchall()]
    return jsonify(results)

@app.route('/history/<int:user_id>', methods=['GET'])
def user_history():
    """View historical user data (time-travel)"""
    timestamp = request.args.get('timestamp')

    conn = get_db()
    # Query historical state
    cursor = conn.execute(f'''
        SELECT * FROM users
        WHERE id = ?
        AS OF TIMESTAMP '{timestamp}'
    ''', (user_id,))

    user = dict(cursor.fetchone() or {})
    return jsonify(user)
```

---

## FastAPI Migration

### Before (FastAPI + SQLite):

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import sqlite3

app = FastAPI()
DATABASE = 'api.db'

class User(BaseModel):
    name: str
    email: str

class UserResponse(User):
    id: int

def get_db():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

@app.get("/users", response_model=list[UserResponse])
async def get_users():
    conn = get_db()
    cursor = conn.execute('SELECT * FROM users')
    users = [dict(row) for row in cursor.fetchall()]
    conn.close()
    return users

@app.post("/users", response_model=UserResponse)
async def create_user(user: User):
    conn = get_db()
    cursor = conn.execute(
        'INSERT INTO users (name, email) VALUES (?, ?)',
        (user.name, user.email)
    )
    user_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return {"id": user_id, **user.dict()}
```

### After (FastAPI + HeliosDB Nano):

```python
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import heliosdb_sqlite as sqlite3  # <- Only change!
from contextlib import contextmanager

app = FastAPI()
DATABASE = 'api.db'

class User(BaseModel):
    name: str
    email: str

class UserResponse(User):
    id: int

@contextmanager
def get_db():
    """Database connection context manager"""
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()

@app.get("/users", response_model=list[UserResponse])
async def get_users():
    with get_db() as conn:
        cursor = conn.execute('SELECT * FROM users')
        users = [dict(row) for row in cursor.fetchall()]
    return users

@app.post("/users", response_model=UserResponse)
async def create_user(user: User):
    with get_db() as conn:
        cursor = conn.execute(
            'INSERT INTO users (name, email) VALUES (?, ?)',
            (user.name, user.email)
        )
        user_id = cursor.lastrowid
    return {"id": user_id, **user.dict()}
```

---

## Asyncio/Concurrent Application Migration

### Handling Concurrent Database Access:

**Before (SQLite with potential locks):**
```python
import sqlite3
import asyncio
from concurrent.futures import ThreadPoolExecutor

DATABASE = 'concurrent.db'

def write_to_db(data):
    """May encounter database locks with SQLite"""
    try:
        conn = sqlite3.connect(DATABASE, timeout=10)
        cursor = conn.cursor()
        cursor.execute('INSERT INTO logs VALUES (?, ?)', data)
        conn.commit()
        conn.close()
        return True
    except sqlite3.OperationalError as e:
        print(f"Lock error: {e}")
        return False

async def concurrent_writes():
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor(max_workers=10) as pool:
        tasks = [
            loop.run_in_executor(pool, write_to_db, (i, f"message_{i}"))
            for i in range(100)
        ]
        results = await asyncio.gather(*tasks)
        print(f"Success rate: {sum(results)}/{len(results)}")

asyncio.run(concurrent_writes())
```

**After (HeliosDB Nano, no locks):**
```python
import heliosdb_sqlite as sqlite3  # <- Only change!
import asyncio
from concurrent.futures import ThreadPoolExecutor

DATABASE = 'concurrent.db'

def write_to_db(data):
    """No locks with HeliosDB Nano!"""
    conn = sqlite3.connect(DATABASE)  # No timeout needed
    cursor = conn.cursor()
    cursor.execute('INSERT INTO logs VALUES (?, ?)', data)
    conn.commit()
    conn.close()
    return True

async def concurrent_writes():
    loop = asyncio.get_event_loop()
    with ThreadPoolExecutor(max_workers=10) as pool:
        tasks = [
            loop.run_in_executor(pool, write_to_db, (i, f"message_{i}"))
            for i in range(100)
        ]
        results = await asyncio.gather(*tasks)
        print(f"Success rate: {sum(results)}/{len(results)}")  # All succeed!

asyncio.run(concurrent_writes())
```

---

## Jupyter Notebook Transition

### Before (Jupyter + SQLite):

```python
# Cell 1: Setup
import sqlite3
import pandas as pd

conn = sqlite3.connect('research.db')

# Cell 2: Create table
conn.execute('''
    CREATE TABLE IF NOT EXISTS experiments (
        id INTEGER PRIMARY KEY,
        name TEXT,
        result REAL
    )
''')
conn.commit()

# Cell 3: Insert data
data = [(1, 'exp1', 0.85), (2, 'exp2', 0.92), (3, 'exp3', 0.78)]
conn.executemany('INSERT OR REPLACE INTO experiments VALUES (?, ?, ?)', data)
conn.commit()

# Cell 4: Query and visualize
df = pd.read_sql_query('SELECT * FROM experiments', conn)
df.plot(x='name', y='result', kind='bar')
```

### After (Jupyter + HeliosDB Nano):

```python
# Cell 1: Setup
import heliosdb_sqlite as sqlite3  # <- Only change!
import pandas as pd

conn = sqlite3.connect('research.db')

# Everything else is IDENTICAL!

# Cell 2: Create table
conn.execute('''
    CREATE TABLE IF NOT EXISTS experiments (
        id INTEGER PRIMARY KEY,
        name TEXT,
        result REAL
    )
''')
conn.commit()

# Cell 3: Insert data
data = [(1, 'exp1', 0.85), (2, 'exp2', 0.92), (3, 'exp3', 0.78)]
conn.executemany('INSERT OR REPLACE INTO experiments VALUES (?, ?, ?)', data)
conn.commit()

# Cell 4: Query and visualize
df = pd.read_sql_query('SELECT * FROM experiments', conn)
df.plot(x='name', y='result', kind='bar')

# Cell 5: NEW - Time-travel to see previous results
df_historical = pd.read_sql_query('''
    SELECT * FROM experiments
    AS OF TIMESTAMP '2024-01-01 10:00:00'
''', conn)
df_historical.plot(x='name', y='result', kind='bar', title='Historical Results')
```

---

## Production Deployment Patterns

### Pattern 1: Environment-Based Configuration

```python
# config.py
import os

DATABASE_CONFIG = {
    'development': {
        'engine': 'heliosdb',
        'path': 'dev.db',
        'mode': 'hybrid',
    },
    'testing': {
        'engine': 'sqlite',
        'path': ':memory:',
    },
    'production': {
        'engine': 'heliosdb',
        'path': '/var/lib/myapp/production.db',
        'mode': 'server',
        'port': 5432,
        'encryption_key': os.getenv('DB_ENCRYPTION_KEY'),
    }
}

def get_database_connection():
    env = os.getenv('ENVIRONMENT', 'development')
    config = DATABASE_CONFIG[env]

    if config['engine'] == 'heliosdb':
        import heliosdb_sqlite as sqlite3
        return sqlite3.connect(
            config['path'],
            heliosdb_mode=config.get('mode', 'advanced'),
            encryption_key=config.get('encryption_key')
        )
    else:
        import sqlite3
        return sqlite3.connect(config['path'])

# Usage
conn = get_database_connection()
```

### Pattern 2: Docker Deployment

```dockerfile
# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install HeliosDB Nano
RUN pip install "git+https://github.com/HeliosDatabase/HeliosDB-SDKs.git#subdirectory=sdks/python-sqlite"

# Copy application
COPY . /app

# Install dependencies
RUN pip install -r requirements.txt

# Expose port (if using server mode)
EXPOSE 5432

# Run application
CMD ["python", "app.py"]
```

```yaml
# docker-compose.yml
version: '3.8'

services:
  app:
    build: .
    ports:
      - "8000:8000"
      - "5432:5432"  # HeliosDB Nano server mode
    volumes:
      - ./data:/var/lib/myapp
    environment:
      - ENVIRONMENT=production
      - DB_ENCRYPTION_KEY=${DB_ENCRYPTION_KEY}
      - HELIOSDB_MODE=server
```

### Pattern 3: Gradual Rollout

```python
# gradual_migration.py
import os
import random

def get_database_module():
    """
    Gradually roll out HeliosDB Nano to users
    """
    # Get rollout percentage from environment
    rollout_percentage = int(os.getenv('HELIOSDB_ROLLOUT', '0'))

    # Use user ID or session ID for consistent experience
    user_id = get_current_user_id()
    use_heliosdb = (hash(user_id) % 100) < rollout_percentage

    if use_heliosdb:
        import heliosdb_sqlite as sqlite3
        log_metric('database_engine', 'heliosdb')
    else:
        import sqlite3
        log_metric('database_engine', 'sqlite')

    return sqlite3

# Usage
sqlite3 = get_database_module()
conn = sqlite3.connect('app.db')
```

### Pattern 4: Blue-Green Deployment

```bash
#!/bin/bash
# deploy_heliosdb.sh

# Step 1: Deploy HeliosDB Nano version to green environment
docker-compose -f docker-compose.green.yml up -d

# Step 2: Run health checks
./health_check.sh green

# Step 3: Switch traffic to green
./switch_traffic.sh green

# Step 4: Monitor for issues
./monitor.sh --duration 300

# Step 5: If successful, shut down blue environment
docker-compose -f docker-compose.blue.yml down
```

---

## Summary

All migration patterns follow the same principle:

**Change one import line, keep everything else the same.**

```python
# The only change in most cases:
import heliosdb_sqlite as sqlite3
```

Then optionally explore advanced features when you're ready:
- Vector search
- Time-travel queries
- Database branching
- Server mode
- Encryption at rest

For more details on advanced features, see `HELIOSDB_SQLITE_ADVANCED_FEATURES.md`.