SQLAlchemy Integration Architecture

This document describes the architecture of dqlitepy's SQLAlchemy dialect implementation, providing ORM capabilities for distributed SQL operations.

Overview

The SQLAlchemy integration provides a custom dialect that bridges SQLAlchemy's ORM layer with dqlite's distributed database engine through dqlitepy's DB-API 2.0 interface.

Dialect Components

DQLiteDialect Class

The dialect class defines dqlite-specific behavior and capabilities:

Key Features

Supported Operations:

• ✅ Table creation/dropping (CREATE TABLE, DROP TABLE)
• ✅ CRUD operations (INSERT, SELECT, UPDATE, DELETE)
• ✅ Transactions (BEGIN, COMMIT, ROLLBACK)
• ✅ Joins (INNER, LEFT, RIGHT, FULL OUTER)
• ✅ Aggregations (COUNT, SUM, AVG, MIN, MAX)
• ✅ Subqueries and CTEs (Common Table Expressions)
• ✅ Indexes (CREATE INDEX, DROP INDEX)
• ✅ Constraints (PRIMARY KEY, FOREIGN KEY, UNIQUE, NOT NULL)
• ✅ Schema introspection (table/column metadata)

Limitations:

• ❌ Sequences (AUTOINCREMENT supported via INTEGER PRIMARY KEY)
• ❌ Server-side cursors (all results buffered client-side)
• ❌ Stored procedures
• ❌ Custom functions (SQLite built-ins available)

Connection Flow

Engine Creation and Connection

URL Format

The dialect uses a custom URL format:

dqlite:///[address]/[data_dir]/[database]

Examples:
- dqlite:///127.0.0.1:9001/tmp/dqlite/app.db
- dqlite:///192.168.1.10:9001/var/lib/dqlite/prod.db

Components:

• address: IP:port for dqlite node (e.g., 127.0.0.1:9001)
• data_dir: Directory for Raft logs and snapshots
• database: Database name (file in data_dir)

Query Execution Flow

SELECT Query

Example:

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

engine = create_engine("dqlite:///127.0.0.1:9001/tmp/dqlite/app.db")

with Session(engine) as session:
    # ORM Query
    users = session.query(User).filter(User.age > 18).all()
    
    # Core Query
    stmt = select(User).where(User.age > 18)
    users = session.execute(stmt).scalars().all()

Generated SQL:

SELECT users.id, users.name, users.age, users.email
FROM users
WHERE users.age > ?

INSERT Query

Example:

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Create new user
    user = User(name="Alice", age=25, email="alice@example.com")
    session.add(user)
    session.commit()
    
    print(f"Created user with ID: {user.id}")

Generated SQL:

INSERT INTO users (name, age, email)
VALUES (?, ?, ?)

UPDATE Query

Example:

with Session(engine) as session:
    # Update existing user
    user = session.query(User).filter(User.id == 1).first()
    user.email = "newemail@example.com"
    session.commit()

Generated SQL:

UPDATE users
SET email = ?
WHERE users.id = ?

Type Mapping

Python ↔ SQLAlchemy ↔ SQLite

Type Compiler Implementation

The DQLiteTypeCompiler handles type conversion:

class DQLiteTypeCompiler(SQLiteTypeCompiler):
    """Type compiler for dqlite dialect."""
    
    def visit_BOOLEAN(self, type_, **kw):
        """Boolean stored as INTEGER (0/1)."""
        return "INTEGER"
    
    def visit_TEXT(self, type_, **kw):
        """Text with optional length."""
        if type_.length:
            return f"TEXT({type_.length})"
        return "TEXT"
    
    def visit_DATETIME(self, type_, **kw):
        """DateTime stored as TEXT in ISO 8601 format."""
        return "TEXT"
    
    def visit_BLOB(self, type_, **kw):
        """Binary data stored as BLOB."""
        return "BLOB"

Schema Introspection

Table Discovery

Example:

from sqlalchemy import inspect

inspector = inspect(engine)

# List all tables
tables = inspector.get_table_names()
print(f"Tables: {tables}")

# Get columns for a table
columns = inspector.get_columns("users")
for col in columns:
    print(f"Column: {col['name']} ({col['type']})")

# Get primary key
pk = inspector.get_pk_constraint("users")
print(f"Primary key: {pk['constrained_columns']}")

# Get foreign keys
fks = inspector.get_foreign_keys("users")
for fk in fks:
    print(f"FK: {fk['constrained_columns']} -> {fk['referred_table']}")

# Get indexes
indexes = inspector.get_indexes("users")
for idx in indexes:
    print(f"Index: {idx['name']} on {idx['column_names']}")

Transaction Management

Transaction Lifecycle

Transaction Isolation

dqlite provides Serializable isolation through Raft consensus:

Transaction Example:

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Explicit transaction
    with session.begin():
        user = User(name="Alice")
        session.add(user)
        
        post = Post(user_id=user.id, title="Hello")
        session.add(post)
        
        # Automatically committed if no exception
    
    # Implicit transaction (autocommit disabled)
    user = session.query(User).filter(User.id == 1).first()
    user.name = "Bob"
    session.commit()  # Explicit commit

ORM Model Definition

Declarative Base

from sqlalchemy import create_engine, Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    posts = relationship("Post", back_populates="author")
    
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}')>"

class Post(Base):
    __tablename__ = "posts"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    published = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationships
    author = relationship("User", back_populates="posts")
    
    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title}')>"

# Create engine
engine = create_engine("dqlite:///127.0.0.1:9001/tmp/dqlite/app.db")

# Create tables
Base.metadata.create_all(engine)

Relationship Queries

Eager Loading Strategies

Examples:

from sqlalchemy.orm import joinedload, subqueryload, selectinload

with Session(engine) as session:
    # Lazy loading (default) - N+1 queries
    users = session.query(User).all()
    for user in users:
        print(user.posts)  # Separate query for each user
    
    # Joined loading - 1 query with JOIN
    users = session.query(User).options(joinedload(User.posts)).all()
    
    # Subquery loading - 2 queries
    users = session.query(User).options(subqueryload(User.posts)).all()
    
    # Select-in loading - 2 queries (recommended)
    users = session.query(User).options(selectinload(User.posts)).all()

Generated SQL (Select-In):

-- Query 1: Get users
SELECT users.id, users.name, users.email
FROM users;

-- Query 2: Get posts for those users
SELECT posts.id, posts.title, posts.user_id
FROM posts
WHERE posts.user_id IN (?, ?, ?);

Connection Pooling

Pool Configuration

Configuration:

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    "dqlite:///127.0.0.1:9001/tmp/dqlite/app.db",
    poolclass=QueuePool,
    pool_size=10,              # Base pool size
    max_overflow=20,           # Additional connections
    pool_timeout=30,           # Seconds to wait for connection
    pool_recycle=3600,         # Recycle connections after 1 hour
    pool_pre_ping=True,        # Verify connections before use
    echo=False,                # Don't log SQL (production)
    echo_pool=False,           # Don't log pool events
)

Performance Optimization

Query Performance Tips

Best Practices:

1. Create Indexes:

from sqlalchemy import Index

# Add index to model
class User(Base):
    __tablename__ = "users"
    
    email = Column(String(255), unique=True, index=True)
    
    __table_args__ = (
        Index("idx_user_email", "email"),
    )

1. Use Pagination:

# Bad - loads everything
users = session.query(User).all()

# Good - paginate
users = session.query(User).limit(100).offset(0).all()

1. Use Bulk Operations:

# Bad - individual inserts
for data in user_data:
    session.add(User(**data))
session.commit()

# Good - bulk insert
session.bulk_insert_mappings(User, user_data)
session.commit()

Error Handling

Dialect-Specific Errors

Example:

from sqlalchemy.exc import (
    IntegrityError,
    OperationalError,
    ProgrammingError
)
from sqlalchemy.orm import Session

with Session(engine) as session:
    try:
        user = User(email="duplicate@example.com")
        session.add(user)
        session.commit()
    except IntegrityError as e:
        session.rollback()
        print(f"Duplicate email: {e}")
    except OperationalError as e:
        session.rollback()
        print(f"Database error: {e}")
    except ProgrammingError as e:
        session.rollback()
        print(f"SQL error: {e}")

Migration Support

Alembic Integration

dqlitepy's SQLAlchemy dialect works with Alembic for schema migrations:

# alembic/env.py
from alembic import context
from sqlalchemy import engine_from_config, pool
from myapp.models import Base

config = context.config
target_metadata = Base.metadata

def run_migrations_online():
    """Run migrations in 'online' mode."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )
        
        with context.begin_transaction():
            context.run_migrations()

alembic.ini:

[alembic]
sqlalchemy.url = dqlite:///127.0.0.1:9001/tmp/dqlite/app.db

Creating Migrations:

# Initialize Alembic
uv run alembic init alembic

# Create migration
uv run alembic revision --autogenerate -m "Add users table"

# Apply migration
uv run alembic upgrade head

# Rollback migration
uv run alembic downgrade -1

Summary

Architecture Highlights

Component	Purpose	Implementation
DQLiteDialect	SQLAlchemy dialect registration	Inherits from SQLiteDialect
DQLiteCompiler	SQL statement compilation	Overrides SQLite compiler
DQLiteTypeCompiler	Type mapping	Maps Python ↔ SQLite types
Connection Pooling	Resource management	QueuePool with pre-ping
Schema Introspection	Metadata discovery	PRAGMA queries
Transaction Support	ACID guarantees	Raft consensus

Key Benefits

• ✅ Familiar ORM Interface: Use standard SQLAlchemy patterns
• ✅ Distributed Consistency: Raft-based replication
• ✅ Schema Migration: Alembic compatibility
• ✅ Type Safety: Pythonic type mapping
• ✅ Connection Pooling: Efficient resource usage
• ✅ Relationship Support: Eager/lazy loading strategies

Performance Characteristics

• Latency: +1-2ms overhead vs direct DB-API
• Throughput: ~1000 queries/sec per connection
• Memory: Object hydration adds ~2-5x memory vs tuples
• Scalability: Horizontal scaling via cluster

References

• SQLAlchemy Documentation
• Alembic Documentation
• dqlitepy DB-API Guide
• dqlitepy Architecture
• FastAPI Integration