SQLAlchemy ORM Example

Learn how to use SQLAlchemy's powerful ORM with dqlitepy for distributed database operations with a Pythonic interface.

Overview

This example demonstrates:

• Configuring SQLAlchemy with dqlitepy
• Defining ORM models
• CRUD operations with the ORM
• Relationships and joins
• Session management

Perfect for: Developers who prefer ORM patterns over raw SQL.

Prerequisites

• Python 3.12 or higher
• uv package manager installed
• dqlitepy installed in your project
• Basic understanding of SQLAlchemy
• Understanding of dqlitepy basics (see Simple Node)

Quick Start

Run the example with one command:

cd examples/sqlalchemy_orm
./quickstart.sh

The script will:

1. Install dqlitepy and SQLAlchemy
2. Set up the ORM models
3. Demonstrate CRUD operations
4. Show relationships and queries

Manual Installation

To run manually:

cd examples/sqlalchemy_orm
uv sync
uv run python -m sqlalchemy_orm_example.main

Architecture

Code Walkthrough

Setting Up the Engine

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, Session
import tempfile

# Create temporary directory for dqlite node
data_dir = tempfile.mkdtemp(prefix="dqlite_sqlalchemy_")

# Create SQLAlchemy engine with dqlitepy dialect
engine = create_engine(
    "dqlite+pydqlite:///myapp.db",
    connect_args={
        "node_id": 1,
        "address": "127.0.0.1:9001",
        "data_dir": data_dir
    }
)

# Create base class for models
Base = declarative_base()

Key Points:

• Use the dqlite+pydqlite:// dialect URL
• Pass node configuration in connect_args
• The engine handles connection pooling automatically

Defining Models

from sqlalchemy import Column, Integer, String, Float, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationship to orders
    orders = relationship("Order", back_populates="user")
    
    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

class Order(Base):
    __tablename__ = "orders"
    
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    product_name = Column(String(100), nullable=False)
    quantity = Column(Integer, default=1)
    price = Column(Float, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # Relationship to user
    user = relationship("User", back_populates="orders")
    
    def __repr__(self):
        return f"<Order(id={self.id}, product='{self.product_name}')>"

Key Points:

• Define models as classes inheriting from Base
• Use SQLAlchemy column types
• Define relationships with relationship()
• Use ForeignKey for table relationships

Creating Tables

# Create all tables
Base.metadata.create_all(engine)
print("Tables created successfully!")

Key Points:

• create_all() creates all defined tables
• Tables are created in the dqlite database
• Idempotent operation (safe to run multiple times)

Creating Records

# Create a session
session = Session(engine)

# Create users
user1 = User(username="alice", email="alice@example.com")
user2 = User(username="bob", email="bob@example.com")

# Add to session
session.add(user1)
session.add(user2)

# Commit to database
session.commit()

print(f"Created users: {user1.id}, {user2.id}")

Key Points:

• Create instances of model classes
• Use session.add() to stage changes
• Call session.commit() to persist
• IDs are auto-generated after commit

Creating Related Records

# Create orders for users
order1 = Order(
    user_id=user1.id,
    product_name="Laptop",
    quantity=1,
    price=999.99
)

order2 = Order(
    user_id=user1.id,
    product_name="Mouse",
    quantity=2,
    price=29.99
)

order3 = Order(
    user_id=user2.id,
    product_name="Keyboard",
    quantity=1,
    price=79.99
)

session.add_all([order1, order2, order3])
session.commit()

print(f"Created {session.query(Order).count()} orders")

Key Points:

• Use add_all() for multiple objects
• Foreign keys establish relationships
• Relationships are enforced by the database

Querying Records

# Query all users
users = session.query(User).all()
print("\nAll Users:")
for user in users:
    print(f"  {user}")

# Query with filter
alice = session.query(User).filter_by(username="alice").first()
print(f"\nFound user: {alice}")

# Query with relationships
user_with_orders = session.query(User).filter_by(id=alice.id).first()
print(f"\nAlice's orders:")
for order in user_with_orders.orders:
    print(f"  {order.product_name}: ${order.price} (qty: {order.quantity})")

Key Points:

• Use query() to build queries
• filter_by() for simple equality filters
• first() returns one result or None
• all() returns a list of results
• Relationships are loaded automatically

Advanced Queries

from sqlalchemy import func

# Aggregate query
total_spent = session.query(
    func.sum(Order.price * Order.quantity)
).filter(Order.user_id == alice.id).scalar()

print(f"\nAlice's total spending: ${total_spent:.2f}")

# Join query
results = session.query(User, Order).join(Order).all()
print("\nAll users with their orders:")
for user, order in results:
    print(f"  {user.username}: {order.product_name}")

# Group by query
user_order_counts = session.query(
    User.username,
    func.count(Order.id).label("order_count")
).join(Order).group_by(User.username).all()

print("\nOrder counts by user:")
for username, count in user_order_counts:
    print(f"  {username}: {count} orders")

Key Points:

• Use func for aggregate functions
• join() for combining tables
• scalar() returns a single value
• group_by() for aggregation

Updating Records

# Update a single record
alice.email = "alice.new@example.com"
session.commit()

# Update with query
session.query(Order).filter_by(
    product_name="Mouse"
).update({"quantity": 3})
session.commit()

print("Records updated!")

Key Points:

• Modify object attributes directly
• Use update() for bulk updates
• Always commit changes

Deleting Records

# Delete a single record
session.delete(order3)
session.commit()

# Delete with query
session.query(Order).filter(Order.quantity < 2).delete()
session.commit()

print("Records deleted!")

Key Points:

• Use delete() on session for single objects
• Use .delete() on query for bulk deletes
• Commit to persist deletions

Session Management

# Context manager pattern
from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

def create_user(username: str, email: str):
    with SessionLocal() as session:
        user = User(username=username, email=email)
        session.add(user)
        session.commit()
        session.refresh(user)  # Load generated ID
        return user

# Use the function
new_user = create_user("charlie", "charlie@example.com")
print(f"Created user: {new_user}")

Key Points:

• Use sessionmaker for session factory
• Context manager ensures cleanup
• refresh() loads server-side values

Expected Output

Setting up SQLAlchemy with dqlitepy...
Tables created successfully!

Creating users...
Created users: 1, 2

Creating orders...
Created 3 orders

All Users:
  <User(id=1, username='alice')>
  <User(id=2, username='bob')>

Found user: <User(id=1, username='alice')>

Alice's orders:
  Laptop: $999.99 (qty: 1)
  Mouse: $29.99 (qty: 2)

Alice's total spending: $1059.97

All users with their orders:
  alice: Laptop
  alice: Mouse
  bob: Keyboard

Order counts by user:
  alice: 2 orders
  bob: 1 orders

Records updated!
Records deleted!

Example completed successfully!

Using with Clusters

To use SQLAlchemy with a dqlite cluster:

from dqlitepy.sqlalchemy import create_cluster_engine

# Create engine connected to cluster
engine = create_cluster_engine(
    database="myapp.db",
    cluster=[
        "127.0.0.1:9001",
        "127.0.0.1:9002",
        "127.0.0.1:9003"
    ]
)

# Use normally
Base.metadata.create_all(engine)
session = Session(engine)

Key Points:

• Use create_cluster_engine() for clusters
• Provide list of node addresses
• Client handles leader discovery automatically

Best Practices

Use Context Managers

def get_user(user_id: int):
    with SessionLocal() as session:
        return session.query(User).filter_by(id=user_id).first()

Eager Loading

from sqlalchemy.orm import joinedload

# Load user with orders in one query
user = session.query(User).options(
    joinedload(User.orders)
).filter_by(id=1).first()

Bulk Operations

# Bulk insert
session.bulk_save_objects([
    User(username=f"user{i}", email=f"user{i}@example.com")
    for i in range(100)
])
session.commit()

Error Handling

from sqlalchemy.exc import IntegrityError

try:
    session.add(User(username="alice", email="duplicate@example.com"))
    session.commit()
except IntegrityError:
    session.rollback()
    print("Username already exists!")

Common Issues

Unique Constraint Violations

Handle duplicates gracefully:

from sqlalchemy.exc import IntegrityError

try:
    session.commit()
except IntegrityError as e:
    session.rollback()
    if "UNIQUE constraint failed" in str(e):
        print("Record already exists")

Lazy Loading Issues

Avoid the N+1 query problem:

# Bad: N+1 queries
users = session.query(User).all()
for user in users:
    print(user.orders)  # Separate query for each user

# Good: Single query with join
users = session.query(User).options(joinedload(User.orders)).all()
for user in users:
    print(user.orders)  # Already loaded

Session Lifecycle

Always close sessions:

session = Session(engine)
try:
    # Your operations
    pass
finally:
    session.close()

Source Code

The complete source code is available at:

• examples/sqlalchemy_orm/sqlalchemy_orm_example/main.py

Next Steps

After mastering SQLAlchemy with dqlitepy:

1. FastAPI Integration - Build REST APIs with SQLAlchemy
2. SQLAlchemy Integration Architecture
3. Advanced Clustering

Related Documentation

• SQLAlchemy API Reference
• DB-API Reference
• SQLAlchemy Official Documentation