Python SQLite with Vectors

Working with SQLite databases and vector extensions for embeddings and semantic search in Python.

Overview

SQLite provides a lightweight, serverless database with Python's built-in sqlite3 module. Vector extensions like sqlite-vec and sqlite-vss add support for storing embeddings and performing similarity searches, enabling semantic search capabilities without external infrastructure.

⤢ zoom

Basic Operations

Core SQLite operations for database connections and query execution.

Key Concepts

• Connection: Database file handle; use :memory: for in-memory databases
• Cursor: Execute queries and fetch results
• Transactions: Auto-commit disabled by default; explicit commit() required
• Context managers: Automatically handle connection cleanup

Common Patterns

import sqlite3

# Basic connection
conn = sqlite3.connect('database.db')
cursor = conn.cursor()

# Execute and commit
cursor.execute("INSERT INTO users (name) VALUES (?)", ('Alice',))
conn.commit()

# Fetch results
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()  # or fetchone(), fetchmany(n)

# Close connection
conn.close()

Examples

# Context manager pattern (recommended)
with sqlite3.connect('database.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
    user = cursor.fetchone()
    # Auto-commits on exit if no exception

# Row factory for dict-like access
conn = sqlite3.connect('database.db')
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute("SELECT * FROM users")
row = cursor.fetchone()
print(row['name'])  # Access by column name

# Executemany for batch inserts
data = [('Alice',), ('Bob',), ('Charlie',)]
cursor.executemany("INSERT INTO users (name) VALUES (?)", data)
conn.commit()

# Transaction control
try:
    cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE id = 1")
    cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE id = 2")
    conn.commit()
except sqlite3.Error:
    conn.rollback()
    raise

Schema Management

Creating and modifying database tables and structures.

Key Concepts

• CREATE TABLE: Define table structure with columns and constraints
• ALTER TABLE: Modify existing tables (limited operations in SQLite)
• Indexes: Improve query performance on specific columns
• BLOB type: Store binary data including vector embeddings

Common Patterns

# Create table with various column types
cursor.execute("""
    CREATE TABLE IF NOT EXISTS documents (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        content TEXT,
        embedding BLOB,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# Add index for faster queries
cursor.execute("CREATE INDEX IF NOT EXISTS idx_title ON documents(title)")

# Check if table exists
cursor.execute("""
    SELECT name FROM sqlite_master
    WHERE type='table' AND name='documents'
""")
exists = cursor.fetchone() is not None

Examples

# Complete schema setup for vector storage
def setup_database(db_path: str):
    with sqlite3.connect(db_path) as conn:
        cursor = conn.cursor()

        # Main documents table
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                content TEXT NOT NULL,
                metadata TEXT,  -- JSON string
                embedding BLOB,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)

        # Full-text search table
        cursor.execute("""
            CREATE VIRTUAL TABLE IF NOT EXISTS documents_fts
            USING fts5(title, content, content=documents, content_rowid=id)
        """)

        # Triggers for FTS sync
        cursor.execute("""
            CREATE TRIGGER IF NOT EXISTS documents_ai AFTER INSERT ON documents BEGIN
                INSERT INTO documents_fts(rowid, title, content)
                VALUES (new.id, new.title, new.content);
            END
        """)

        conn.commit()

# Add column (SQLite limitation: no DROP COLUMN before 3.35)
cursor.execute("ALTER TABLE documents ADD COLUMN source TEXT")

# Rename table
cursor.execute("ALTER TABLE old_name RENAME TO new_name")

# Get table schema
cursor.execute("PRAGMA table_info(documents)")
columns = cursor.fetchall()
for col in columns:
    print(f"{col[1]}: {col[2]}")  # name: type

Vector Extension Setup

Installing and loading vector extensions for embedding storage and similarity search.

Key Concepts

• sqlite-vec: Current, actively maintained vector extension — use this
• sqlite-vss: Superseded predecessor; no longer actively maintained; use sqlite-vec instead
• Extension loading: Load via conn.load_extension() or enable_load_extension()
• Vector dimensions: Must be consistent across all embeddings

⤢ zoom

Common Patterns

# sqlite-vec installation and setup
# pip install sqlite-vec

import sqlite3
import sqlite_vec

# Load sqlite-vec extension
conn = sqlite3.connect('vectors.db')
conn.enable_load_extension(True)
sqlite_vec.load(conn)
conn.enable_load_extension(False)

# Create vector table (sqlite-vec)
cursor = conn.cursor()
cursor.execute("""
    CREATE VIRTUAL TABLE IF NOT EXISTS vec_documents
    USING vec0(
        document_id INTEGER PRIMARY KEY,
        embedding FLOAT[384]
    )
""")

Examples

# Complete sqlite-vec setup with error handling
import sqlite3
from pathlib import Path

def create_vector_db(db_path: str, dimensions: int = 384):
    """Create database with vector support."""
    try:
        import sqlite_vec
    except ImportError:
        raise ImportError("Install sqlite-vec: pip install sqlite-vec")

    conn = sqlite3.connect(db_path)
    conn.enable_load_extension(True)
    sqlite_vec.load(conn)
    conn.enable_load_extension(False)

    cursor = conn.cursor()

    # Metadata table
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            content TEXT NOT NULL,
            metadata TEXT
        )
    """)

    # Vector table with specified dimensions
    cursor.execute(f"""
        CREATE VIRTUAL TABLE IF NOT EXISTS vec_documents
        USING vec0(
            document_id INTEGER PRIMARY KEY,
            embedding FLOAT[{dimensions}]
        )
    """)

    conn.commit()
    return conn

# sqlite-vss is the predecessor to sqlite-vec. It is unmaintained, has no
# aarch64 wheels, and should not be used for new work. Use sqlite-vec instead.

# Check extension loaded correctly
cursor.execute("SELECT vec_version()")
version = cursor.fetchone()[0]
print(f"sqlite-vec version: {version}")

Vector Operations

Inserting, updating, and querying vector embeddings.

Key Concepts

• Serialisation: Convert numpy arrays to bytes for storage
• Distance metrics: L2 (Euclidean), cosine similarity
• KNN search: Find K nearest neighbours to query vector
• Batch operations: Insert multiple vectors efficiently

Common Patterns

import numpy as np
import struct

# Serialise vector to bytes
def serialise_vector(vector: list | np.ndarray) -> bytes:
    """Convert vector to bytes for SQLite storage."""
    if isinstance(vector, np.ndarray):
        vector = vector.tolist()
    return struct.pack(f'{len(vector)}f', *vector)

# Deserialise bytes to vector
def deserialise_vector(blob: bytes) -> list:
    """Convert bytes back to vector."""
    n = len(blob) // 4  # 4 bytes per float
    return list(struct.unpack(f'{n}f', blob))

# Insert vector (sqlite-vec)
embedding = [0.1, 0.2, 0.3, ...]  # Your embedding
cursor.execute(
    "INSERT INTO vec_documents (document_id, embedding) VALUES (?, ?)",
    (doc_id, serialise_vector(embedding))
)

Examples

import numpy as np
import struct
import json

class VectorStore:
    """SQLite-based vector store with sqlite-vec."""

    def __init__(self, db_path: str, dimensions: int = 384):
        import sqlite_vec

        self.conn = sqlite3.connect(db_path)
        self.conn.enable_load_extension(True)
        sqlite_vec.load(self.conn)
        self.conn.enable_load_extension(False)
        self.dimensions = dimensions
        self._setup_tables()

    def _setup_tables(self):
        cursor = self.conn.cursor()
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS documents (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                content TEXT NOT NULL,
                metadata TEXT
            )
        """)
        cursor.execute(f"""
            CREATE VIRTUAL TABLE IF NOT EXISTS vec_documents
            USING vec0(document_id INTEGER PRIMARY KEY, embedding FLOAT[{self.dimensions}])
        """)
        self.conn.commit()

    def add_document(self, content: str, embedding: list, metadata: dict = None):
        """Add document with embedding."""
        cursor = self.conn.cursor()

        # Insert document
        cursor.execute(
            "INSERT INTO documents (content, metadata) VALUES (?, ?)",
            (content, json.dumps(metadata) if metadata else None)
        )
        doc_id = cursor.lastrowid

        # Insert embedding
        blob = struct.pack(f'{len(embedding)}f', *embedding)
        cursor.execute(
            "INSERT INTO vec_documents (document_id, embedding) VALUES (?, ?)",
            (doc_id, blob)
        )

        self.conn.commit()
        return doc_id

    def add_documents_batch(self, documents: list[tuple[str, list, dict]]):
        """Batch insert documents with embeddings."""
        cursor = self.conn.cursor()

        for content, embedding, metadata in documents:
            cursor.execute(
                "INSERT INTO documents (content, metadata) VALUES (?, ?)",
                (content, json.dumps(metadata) if metadata else None)
            )
            doc_id = cursor.lastrowid

            blob = struct.pack(f'{len(embedding)}f', *embedding)
            cursor.execute(
                "INSERT INTO vec_documents (document_id, embedding) VALUES (?, ?)",
                (doc_id, blob)
            )

        self.conn.commit()

    def search(self, query_embedding: list, k: int = 10) -> list[dict]:
        """Find k nearest neighbours."""
        cursor = self.conn.cursor()

        blob = struct.pack(f'{len(query_embedding)}f', *query_embedding)

        # sqlite-vec requires LIMIT on the vec0 table directly;
        # use a subquery when joining with other tables
        cursor.execute("""
            SELECT
                d.id,
                d.content,
                d.metadata,
                v.distance
            FROM (
                SELECT document_id, distance
                FROM vec_documents
                WHERE embedding MATCH ?
                LIMIT ?
            ) v
            JOIN documents d ON d.id = v.document_id
            ORDER BY v.distance
        """, (blob, k))

        results = []
        for row in cursor.fetchall():
            results.append({
                'id': row[0],
                'content': row[1],
                'metadata': json.loads(row[2]) if row[2] else None,
                'distance': row[3]
            })

        return results

    def delete_document(self, doc_id: int):
        """Delete document and its embedding."""
        cursor = self.conn.cursor()
        cursor.execute("DELETE FROM vec_documents WHERE document_id = ?", (doc_id,))
        cursor.execute("DELETE FROM documents WHERE id = ?", (doc_id,))
        self.conn.commit()

    def close(self):
        self.conn.close()

# Usage
store = VectorStore('vectors.db', dimensions=384)

# Add single document
doc_id = store.add_document(
    content="Python is a programming language",
    embedding=[0.1] * 384,  # Replace with actual embedding
    metadata={'source': 'docs'}
)

# Search
results = store.search(query_embedding=[0.1] * 384, k=5)
for result in results:
    print(f"Distance: {result['distance']:.4f} - {result['content'][:50]}")

Index Creation

Creating and managing indexes for efficient vector searches.

Key Concepts

• Flat scan: sqlite-vec v0.x uses brute-force exact search (no ANN index); HNSW is on the roadmap but not yet available
• distance_metric: Choose L2 (default) or cosine in the vec0 constructor
• Rebuild: Re-inserting rows may be needed after bulk loads for consistent results
• Memory usage: Larger vectors and row counts increase memory pressure

⤢ zoom

Common Patterns

# sqlite-vec: vec0 uses a flat (brute-force) scan — no HNSW index in v0.x.
# Specify distance_metric in the constructor; search is always exact.
cursor.execute(f"""
    CREATE VIRTUAL TABLE IF NOT EXISTS vec_documents
    USING vec0(
        document_id INTEGER PRIMARY KEY,
        embedding FLOAT[384] distance_metric=cosine
    )
""")

# Check index statistics
cursor.execute("SELECT * FROM vec_documents_info")
info = cursor.fetchall()

Examples

# Configure index for different use cases

# High accuracy (slower, more memory)
cursor.execute("""
    CREATE VIRTUAL TABLE IF NOT EXISTS vec_high_accuracy
    USING vec0(
        document_id INTEGER PRIMARY KEY,
        embedding FLOAT[384] distance_metric=L2
    )
""")

# Cosine similarity (normalised vectors)
cursor.execute("""
    CREATE VIRTUAL TABLE IF NOT EXISTS vec_cosine
    USING vec0(
        document_id INTEGER PRIMARY KEY,
        embedding FLOAT[384] distance_metric=cosine
    )
""")

# Optimise after bulk insert
def bulk_insert_with_optimisation(store, documents):
    """Insert documents and optimise index."""
    # Disable auto-commit for speed
    store.conn.isolation_level = None
    cursor = store.conn.cursor()
    cursor.execute("BEGIN")

    for content, embedding, metadata in documents:
        store.add_document(content, embedding, metadata)

    cursor.execute("COMMIT")
    store.conn.isolation_level = ''

# Check vector count and index health
cursor.execute("SELECT COUNT(*) FROM vec_documents")
count = cursor.fetchone()[0]
print(f"Total vectors: {count}")

Common Use Cases

Practical applications of SQLite vector storage.

Key Concepts

• Semantic search: Find similar content by meaning, not keywords
• RAG (Retrieval-Augmented Generation): Retrieve context for LLM prompts
• Deduplication: Find near-duplicate documents
• Recommendations: Suggest similar items based on embeddings

⤢ zoom

Examples

# Semantic search with sentence-transformers
from sentence_transformers import SentenceTransformer

class SemanticSearch:
    """Semantic search using SQLite vectors."""

    def __init__(self, db_path: str, model_name: str = 'all-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)
        self.dimensions = self.model.get_sentence_embedding_dimension()
        self.store = VectorStore(db_path, self.dimensions)

    def index_documents(self, documents: list[str], metadata: list[dict] = None):
        """Index documents with embeddings."""
        embeddings = self.model.encode(documents)

        batch = []
        for i, (doc, emb) in enumerate(zip(documents, embeddings)):
            meta = metadata[i] if metadata else None
            batch.append((doc, emb.tolist(), meta))

        self.store.add_documents_batch(batch)

    def search(self, query: str, k: int = 5) -> list[dict]:
        """Search for similar documents."""
        query_embedding = self.model.encode(query).tolist()
        return self.store.search(query_embedding, k)

# Usage
search = SemanticSearch('semantic.db')

# Index documents
documents = [
    "Python is great for data science",
    "JavaScript runs in the browser",
    "Machine learning requires lots of data",
    "SQLite is a lightweight database",
]
search.index_documents(documents)

# Search
results = search.search("database for small projects")
for r in results:
    print(f"{r['distance']:.4f}: {r['content']}")


# RAG context retrieval
class RAGRetriever:
    """Retrieve context for RAG applications."""

    def __init__(self, db_path: str):
        self.search = SemanticSearch(db_path)

    def get_context(self, query: str, k: int = 3, max_tokens: int = 1000) -> str:
        """Retrieve relevant context for a query."""
        results = self.search.search(query, k=k)

        context_parts = []
        total_chars = 0

        for result in results:
            content = result['content']
            if total_chars + len(content) > max_tokens * 4:  # Rough char estimate
                break
            context_parts.append(content)
            total_chars += len(content)

        return "\n\n".join(context_parts)

# Usage
retriever = RAGRetriever('knowledge.db')
context = retriever.get_context("How do I connect to SQLite?")
prompt = f"""Based on the following context, answer the question.

Context:
{context}

Question: How do I connect to SQLite?
"""


# Document deduplication
def find_duplicates(store: VectorStore, threshold: float = 0.1) -> list[tuple[int, int]]:
    """Find near-duplicate documents."""
    cursor = store.conn.cursor()
    cursor.execute("SELECT id, content FROM documents")
    documents = cursor.fetchall()

    duplicates = []
    checked = set()

    for doc_id, content in documents:
        if doc_id in checked:
            continue

        # Get embedding for this document
        cursor.execute(
            "SELECT embedding FROM vec_documents WHERE document_id = ?",
            (doc_id,)
        )
        embedding = cursor.fetchone()[0]

        # Search for similar — sqlite-vec does not support extra WHERE
        # filters alongside MATCH; filter in Python after the query
        cursor.execute("""
            SELECT document_id, distance
            FROM vec_documents
            WHERE embedding MATCH ?
            ORDER BY distance
            LIMIT 11
        """, (embedding,))
        # Skip self-match in Python
        similar_rows = [(sid, dist) for sid, dist in cursor.fetchall() if sid != doc_id][:10]

        for similar_id, distance in similar_rows:
            if distance < threshold and similar_id not in checked:
                duplicates.append((doc_id, similar_id))
                checked.add(similar_id)

        checked.add(doc_id)

    return duplicates


# Hybrid search (vector + keyword)
def hybrid_search(
    conn,
    query: str,
    query_embedding: list,
    k: int = 10,
    vector_weight: float = 0.7
) -> list[dict]:
    """Combine vector similarity with FTS ranking."""
    cursor = conn.cursor()

    blob = struct.pack(f'{len(query_embedding)}f', *query_embedding)

    # Vector search
    cursor.execute("""
        SELECT document_id, distance
        FROM vec_documents
        WHERE embedding MATCH ?
        ORDER BY distance
        LIMIT ?
    """, (blob, k * 2))
    vector_results = {row[0]: row[1] for row in cursor.fetchall()}

    # FTS search
    cursor.execute("""
        SELECT rowid, rank
        FROM documents_fts
        WHERE documents_fts MATCH ?
        ORDER BY rank
        LIMIT ?
    """, (query, k * 2))
    fts_results = {row[0]: -row[1] for row in cursor.fetchall()}  # Negate rank

    # Combine scores
    all_ids = set(vector_results.keys()) | set(fts_results.keys())
    combined = []

    for doc_id in all_ids:
        vec_score = 1 - vector_results.get(doc_id, 1)  # Convert distance to similarity
        fts_score = fts_results.get(doc_id, 0)

        # Normalise and combine
        final_score = (vector_weight * vec_score) + ((1 - vector_weight) * fts_score)
        combined.append((doc_id, final_score))

    combined.sort(key=lambda x: x[1], reverse=True)

    # Fetch documents
    results = []
    for doc_id, score in combined[:k]:
        cursor.execute("SELECT content, metadata FROM documents WHERE id = ?", (doc_id,))
        row = cursor.fetchone()
        results.append({
            'id': doc_id,
            'content': row[0],
            'metadata': json.loads(row[1]) if row[1] else None,
            'score': score
        })

    return results

Quick Reference

Operation	Code
Connect	conn = sqlite3.connect('db.db')
Create cursor	cursor = conn.cursor()
Execute query	cursor.execute("SELECT * FROM t WHERE id = ?", (1,))
Fetch one	row = cursor.fetchone()
Fetch all	rows = cursor.fetchall()
Commit	conn.commit()
Load sqlite-vec	sqlite_vec.load(conn)
Serialise vector	struct.pack(f'{n}f', *vector)
Create vec table	CREATE VIRTUAL TABLE t USING vec0(embedding FLOAT[384])
Vector search	WHERE embedding MATCH ? ORDER BY distance LIMIT k
Insert vector	INSERT INTO vec_t (id, embedding) VALUES (?, ?)

Common Issues and Solutions

Issue	Cause	Solution
no such function: vec_version	Extension not loaded	Call sqlite_vec.load(conn) after enabling extensions
extension loading disabled	Security restriction	Use conn.enable_load_extension(True) before loading
dimension mismatch	Embedding size differs	Ensure all vectors match table definition (e.g., FLOAT[384])
database is locked	Concurrent access	Use WAL mode: PRAGMA journal_mode=WAL
Slow searches	Large dataset	sqlite-vec uses flat scan; reduce dataset size or use chunked queries
Memory errors	Large batch inserts	Use chunked inserts; increase page cache
struct.error	Wrong vector format	Ensure vector is flat list of floats, not nested
Poor search results	Embedding model mismatch	Use same model for indexing and querying

Performance Tips

# Enable WAL mode for concurrent reads
conn.execute("PRAGMA journal_mode=WAL")

# Increase cache size (negative = KB)
conn.execute("PRAGMA cache_size=-64000")  # 64MB

# Synchronous mode for speed (risk: data loss on crash)
conn.execute("PRAGMA synchronous=NORMAL")

# Memory-map for large databases
conn.execute("PRAGMA mmap_size=268435456")  # 256MB

Debugging Vector Operations

# Check vector dimensions
cursor.execute("PRAGMA table_info(vec_documents)")
print(cursor.fetchall())

# Verify embedding stored correctly
cursor.execute("SELECT embedding FROM vec_documents WHERE document_id = ?", (1,))
blob = cursor.fetchone()[0]
vector = struct.unpack(f'{len(blob)//4}f', blob)
print(f"Dimensions: {len(vector)}, First 5: {vector[:5]}")

# Test distance calculation
cursor.execute("""
    SELECT document_id, distance
    FROM vec_documents
    WHERE embedding MATCH ?
    LIMIT 5
""", (blob,))
for row in cursor.fetchall():
    print(f"ID: {row[0]}, Distance: {row[1]}")