Skip to content

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
  2. SQLAlchemy Migration
  3. Raw SQL Application Migration
  4. Flask Application Migration
  5. FastAPI Migration
  6. Asyncio/Concurrent Application Migration
  7. Jupyter Notebook Transition
  8. Production Deployment Patterns

Django ORM Migration

Django applications can use HeliosDB Nano with minimal configuration changes.

Step 1: Install HeliosDB Nano Django Backend

Terminal window
pip install heliosdb-django

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)

Terminal window
# 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 Nano 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 Nano

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)
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 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 Nano):

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

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 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 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 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, 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):

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:

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):

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):

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 Nano - 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 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: 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):

# 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 Nano
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 Nano 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 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

deploy_heliosdb.sh
#!/bin/bash
# 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.

# 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.