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
- 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 Nano with minimal configuration changes.
Step 1: Install HeliosDB Nano Django Backend
pip install heliosdb-djangoStep 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 integritypython manage.py check --database default
# Run your testspython manage.py testStep 4: Leverage Advanced Features (Optional)
from django.db import modelsfrom 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 Nano specific) models.Index(fields=['description_embedding'], name='vec_idx'), ]
# Usage in views.pyfrom 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 Nano
from django.test import TestCasefrom .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) 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 Nano’s SQLite compatibility layer.
Step 1: Update Connection String
Before:
from sqlalchemy import create_engine
engine = create_engine('sqlite:///myapp.db')After:
from sqlalchemy import create_engine
# Option 1: Use HeliosDB Nano dialectengine = create_engine('heliosdb:///myapp.db')
# Option 2: Keep sqlite:// with custom creatorimport heliosdb_sqliteengine = 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, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom 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 engineengine = create_engine('sqlite:///users.db')Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)session = Session()
# Create usernew_user = User(name='Alice', email='alice@example.com')session.add(new_user)session.commit()
# Query usersusers = session.query(User).all()for user in users: print(f"{user.name}: {user.email}")After (HeliosDB Nano):
from sqlalchemy import create_engine, Column, Integer, Stringfrom sqlalchemy.ext.declarative import declarative_basefrom 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 identicalnew_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, Textfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom 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 embeddingdoc = 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 searchquery_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_enginefrom sqlalchemy.pool import QueuePool
# Production configuration with connection poolingengine = 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 sessionsfrom contextlib import contextmanager
@contextmanagerdef get_session(): session = Session() try: yield session session.commit() except Exception: session.rollback() raise finally: session.close()
# Usagewith get_session() as session: user = session.query(User).filter_by(email='alice@example.com').first() user.name = 'Alice Smith' # Automatically committed on context exitRaw 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()
# Usagedb = UserDatabase('users.db')user_id = db.create_user('alice', 'alice@example.com')print(db.get_user(user_id))After (HeliosDB Nano):
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 identicaldb = 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 sqlite3from 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):
from flask import Flask, request, jsonifyimport 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):
from flask import Flask, request, jsonify, gimport 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_appcontextdef 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:
from flask import Flask, request, jsonify, gimport 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):
from fastapi import FastAPI, HTTPExceptionfrom pydantic import BaseModelimport 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):
from fastapi import FastAPI, HTTPException, Dependsfrom pydantic import BaseModelimport 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
@contextmanagerdef 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 sqlite3import asynciofrom 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!):
import heliosdb_sqlite as sqlite3 # <- Only change!import asynciofrom 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):
# Cell 1: Setupimport sqlite3import pandas as pd
conn = sqlite3.connect('research.db')
# Cell 2: Create tableconn.execute(''' CREATE TABLE IF NOT EXISTS experiments ( id INTEGER PRIMARY KEY, name TEXT, result REAL )''')conn.commit()
# Cell 3: Insert datadata = [(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 visualizedf = pd.read_sql_query('SELECT * FROM experiments', conn)df.plot(x='name', y='result', kind='bar')After (Jupyter + HeliosDB Nano):
# Cell 1: Setupimport heliosdb_sqlite as sqlite3 # <- Only change!import pandas as pd
conn = sqlite3.connect('research.db')
# Everything else is IDENTICAL!
# Cell 2: Create tableconn.execute(''' CREATE TABLE IF NOT EXISTS experiments ( id INTEGER PRIMARY KEY, name TEXT, result REAL )''')conn.commit()
# Cell 3: Insert datadata = [(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 visualizedf = pd.read_sql_query('SELECT * FROM experiments', conn)df.plot(x='name', y='result', kind='bar')
# Cell 5: NEW - Time-travel to see previous resultsdf_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
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'])
# Usageconn = get_database_connection()Pattern 2: Docker Deployment
# DockerfileFROM python:3.11-slim
WORKDIR /app
# Install HeliosDB NanoRUN pip install heliosdb-sqlite
# Copy applicationCOPY . /app
# Install dependenciesRUN pip install -r requirements.txt
# Expose port (if using server mode)EXPOSE 5432
# Run applicationCMD ["python", "app.py"]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=serverPattern 3: Gradual Rollout
import osimport 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
# Usagesqlite3 = get_database_module()conn = sqlite3.connect('app.db')Pattern 4: Blue-Green Deployment
#!/bin/bash# Step 1: Deploy HeliosDB Nano version to green environmentdocker-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 environmentdocker-compose -f docker-compose.blue.yml downSummary
All migration patterns follow the same principle:
Change one import line, keep everything else the same.
# The only change in most cases:import heliosdb_sqlite as sqlite3Then 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.