Components Database ORM

SQLAlchemy

Core Stack

Python SQL toolkit and Object Relational Mapping library

Version

2.0.0

Last Updated

2024-01-03

Difficulty

Advanced

Reading Time

4 min

GitHub Documentation PyPI

SQLAlchemy

SQLAlchemy is the Python SQL toolkit and Object Relational Mapping (ORM) library that gives application developers the full power and flexibility of SQL. It provides a full suite of well known enterprise-level persistence patterns.

Key Features

Mature and Feature-Rich: Over 15 years of development and refinement
Excellent Performance: Highly optimized for speed and efficiency
Flexible Architecture: Core and ORM layers can be used independently
Strong Typing Support: Full support for Python type hints
Advanced Query Capabilities: Powerful query construction and execution

Installation

1

pip install sqlalchemy

For async support:

1

pip install sqlalchemy[asyncio]

Quick Start

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Create engine
engine = create_engine('sqlite:///example.db')

# Create base class
Base = declarative_base()

# Define model
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# Create tables
Base.metadata.create_all(engine)

# Create session
Session = sessionmaker(bind=engine)
session = Session()

# Create and save user
user = User(name="John Doe", email="[email protected]")
session.add(user)
session.commit()

SQLAlchemy 2.0 Style

from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    email: Mapped[str] = mapped_column(String(100))

engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)

# Using the new session style
with Session(engine) as session:
    # Create user
    user = User(name="Jane Doe", email="[email protected]")
    session.add(user)
    session.commit()
    
    # Query users
    stmt = select(User).where(User.name == "Jane Doe")
    result = session.execute(stmt)
    user = result.scalar_one()

Use Cases

Complex Database Applications: Handle sophisticated data relationships
Data-Intensive Applications: Optimize for high-volume data operations
Enterprise Applications: Robust features for large-scale systems
Multi-Database Support: Work with different database backends

Best Practices

Use Type Hints: Leverage Mapped types for better IDE support
Session Management: Properly manage database sessions and connections
Query Optimization: Use eager loading to avoid N+1 queries
Connection Pooling: Configure appropriate connection pool settings
Migration Management: Use Alembic for database schema migrations

Common Patterns

Relationships

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = 'users'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    
    posts: Mapped[List["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = 'posts'
    
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    content: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    
    author: Mapped["User"] = relationship(back_populates="posts")

Async Operations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.asyncio import async_sessionmaker

# Create async engine
async_engine = create_async_engine('sqlite+aiosqlite:///example.db')

# Create async session factory
async_session = async_sessionmaker(async_engine)

async def create_user(name: str, email: str):
    async with async_session() as session:
        user = User(name=name, email=email)
        session.add(user)
        await session.commit()
        return user

async def get_user_by_email(email: str):
    async with async_session() as session:
        stmt = select(User).where(User.email == email)
        result = await session.execute(stmt)
        return result.scalar_one_or_none()

FastAPI Integration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41


from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel

app = FastAPI()

# Dependency to get database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Pydantic models
class UserCreate(BaseModel):
    name: str
    email: str

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    
    class Config:
        from_attributes = True

@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):
    user = db.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Query Optimization

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


from sqlalchemy.orm import selectinload, joinedload

# Eager loading to avoid N+1 queries
def get_users_with_posts():
    with Session(engine) as session:
        stmt = select(User).options(selectinload(User.posts))
        users = session.execute(stmt).scalars().all()
        return users

# Joined loading for one-to-one relationships
def get_posts_with_authors():
    with Session(engine) as session:
        stmt = select(Post).options(joinedload(Post.author))
        posts = session.execute(stmt).scalars().all()
        return posts

Migration with Alembic

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


# Install Alembic
pip install alembic

# Initialize Alembic
alembic init alembic

# Create migration
alembic revision --autogenerate -m "Create users table"

# Apply migration
alembic upgrade head

Resources

Alternatives

Django ORM

Peewee

Lightweight Python ORM

Key Strengths:

• Simple and lightweight
• Easy to learn

Best For:

• Small applications
• Prototyping

Beginner

Tortoise ORM

Quick Decision Guide

Choose SQLAlchemy for the recommended stack with proven patterns and comprehensive support.

Choose Peewee if you need small applications or similar specialized requirements.

View detailed comparisons →

#python #database #orm #sql