HeliosDB-Lite SQLite Migration Patterns¶
Real-world examples for migrating common Python applications and frameworks to HeliosDB-Lite.
Table of Contents¶
- Django ORM Migration
- SQLAlchemy Migration
- Raw SQL Application Migration
- Flask Application Migration
- FastAPI Migration
- Asyncio/Concurrent Application Migration
- Jupyter Notebook Transition
- Production Deployment Patterns
Django ORM Migration¶
Django applications can use HeliosDB-Lite with minimal configuration changes.
Step 1: Install HeliosDB-Lite Django Backend¶
Step 2: Update Django Settings¶
Before (settings.py):
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': BASE_DIR / 'db.sqlite3',
}
}
After (settings.py):
DATABASES = {
'default': {
'ENGINE': 'heliosdb_django',
'NAME': BASE_DIR / 'db.sqlite3',
# Optional: Enable advanced features
'OPTIONS': {
'heliosdb_mode': 'hybrid', # or 'server', 'advanced'
'enable_vector_search': True,
'enable_time_travel': True,
}
}
}
Step 3: Run Migrations (Existing Database)¶
# 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)¶
# models.py
from django.db import models
from heliosdb_django.fields import VectorField
class Product(models.Model):
name = models.CharField(max_length=200)
description = models.TextField()
# Add vector search for semantic similarity
description_embedding = VectorField(dimensions=384, null=True)
class Meta:
indexes = [
# Traditional index
models.Index(fields=['name']),
# Vector similarity index (HeliosDB-Lite specific)
models.Index(fields=['description_embedding'], name='vec_idx'),
]
# Usage in views.py
from heliosdb_django.vector import similarity_search
def search_products(request):
query = request.GET.get('q', '')
query_embedding = generate_embedding(query)
# Semantic search using vectors
similar_products = similarity_search(
Product,
'description_embedding',
query_embedding,
limit=10
)
return render(request, 'products.html', {'products': similar_products})
Step 5: Testing Django + HeliosDB-Lite¶
# 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-Lite specific)
from heliosdb_django.time_travel import query_at_timestamp
past_time = timezone.now() - timedelta(seconds=5)
historical_product = query_at_timestamp(
Product.objects.filter(id=product.id),
past_time
).first()
self.assertEqual(historical_product.description, "Original")
SQLAlchemy Migration¶
SQLAlchemy works seamlessly with HeliosDB-Lite's SQLite compatibility layer.
Step 1: Update Connection String¶
Before:
After:
from sqlalchemy import create_engine
# Option 1: Use HeliosDB-Lite 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):
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-Lite):
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-Lite 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¶
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¶
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):¶
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-Lite):¶
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:¶
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-Lite)"""
with self.get_connection() as conn:
conn.executemany(
'INSERT INTO users (username, email) VALUES (?, ?)',
users
)
Flask Application Migration¶
Before (Flask + SQLite):¶
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-Lite):¶
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-Lite Advanced Features:¶
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-Lite 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):¶
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-Lite):¶
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):
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-Lite - No locks!):
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-Lite!"""
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):¶
# 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-Lite):¶
# 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¶
# 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
FROM python:3.11-slim
WORKDIR /app
# Install HeliosDB-Lite
RUN pip install heliosdb-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"]
# docker-compose.yml
version: '3.8'
services:
app:
build: .
ports:
- "8000:8000"
- "5432:5432" # HeliosDB-Lite server mode
volumes:
- ./data:/var/lib/myapp
environment:
- ENVIRONMENT=production
- DB_ENCRYPTION_KEY=${DB_ENCRYPTION_KEY}
- HELIOSDB_MODE=server
Pattern 3: Gradual Rollout¶
# gradual_migration.py
import os
import random
def get_database_module():
"""
Gradually roll out HeliosDB-Lite 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¶
#!/bin/bash
# deploy_heliosdb.sh
# Step 1: Deploy HeliosDB-Lite 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.
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.