Database Queries

Database Queries Guide

A comprehensive guide to writing efficient database queries using SQLAlchemy in your FastAPI Clean Architecture application.

Table of Contents

• SQLAlchemy Basics
• Query Patterns
• Performance Optimization
• Common Problems & Solutions
• Advanced Techniques
• Best Practices
• Examples

SQLAlchemy Basics

Async Session Usage

In our FastAPI template, we use async SQLAlchemy sessions:

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, insert, update, delete
from sqlalchemy.orm import selectinload, joinedload

async def get_user_by_id(session: AsyncSession, user_id: int) -> User | None:
    """Get user by ID with proper async handling."""
    stmt = select(User).where(User.id == user_id)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

Basic Query Operations

SELECT Queries

# Simple select
stmt = select(User)
result = await session.execute(stmt)
users = result.scalars().all()

# Select with conditions
stmt = select(User).where(User.is_active == True)
result = await session.execute(stmt)
active_users = result.scalars().all()

# Select specific columns
stmt = select(User.id, User.email).where(User.created_at > datetime.now() - timedelta(days=30))
result = await session.execute(stmt)
recent_users = result.all()

INSERT Operations

# Single insert
new_user = User(email="user@example.com", username="newuser")
session.add(new_user)
await session.commit()

# Bulk insert
stmt = insert(User).values([
    {"email": "user1@example.com", "username": "user1"},
    {"email": "user2@example.com", "username": "user2"},
])
await session.execute(stmt)
await session.commit()

UPDATE Operations

# Update single record
stmt = update(User).where(User.id == user_id).values(last_login=datetime.now())
await session.execute(stmt)
await session.commit()

# Bulk update
stmt = update(User).where(User.is_active == False).values(deactivated_at=datetime.now())
result = await session.execute(stmt)
print(f"Updated {result.rowcount} users")
await session.commit()

DELETE Operations

# Delete single record
stmt = delete(User).where(User.id == user_id)
await session.execute(stmt)
await session.commit()

# Conditional delete
stmt = delete(User).where(
    User.created_at < datetime.now() - timedelta(days=365),
    User.is_active == False
)
result = await session.execute(stmt)
print(f"Deleted {result.rowcount} inactive users")
await session.commit()

Query Patterns

Filtering and Conditions

from sqlalchemy import and_, or_, not_, func

# Multiple conditions with AND
stmt = select(User).where(
    and_(
        User.is_active == True,
        User.created_at > datetime.now() - timedelta(days=30)
    )
)

# Multiple conditions with OR
stmt = select(User).where(
    or_(
        User.role == "admin",
        User.role == "moderator"
    )
)

# Complex conditions
stmt = select(User).where(
    and_(
        User.is_active == True,
        or_(
            User.last_login > datetime.now() - timedelta(days=7),
            User.role == "premium"
        )
    )
)

# NOT conditions
stmt = select(User).where(
    not_(User.role.in_(["banned", "suspended"]))
)

# LIKE patterns
stmt = select(User).where(User.email.like("%@gmail.com"))

# Case-insensitive search
stmt = select(User).where(func.lower(User.username).like(f"%{search_term.lower()}%"))

Ordering and Limiting

# Simple ordering
stmt = select(User).order_by(User.created_at.desc())

# Multiple order criteria
stmt = select(User).order_by(User.role.asc(), User.created_at.desc())

# Pagination
stmt = select(User).order_by(User.id).limit(20).offset(page * 20)

# Top N records
stmt = select(User).order_by(User.score.desc()).limit(10)

Aggregations

from sqlalchemy import func, distinct

# Count records
stmt = select(func.count(User.id)).where(User.is_active == True)
result = await session.execute(stmt)
active_count = result.scalar()

# Group by with aggregation
stmt = select(
    User.role,
    func.count(User.id).label('user_count'),
    func.avg(User.score).label('avg_score')
).group_by(User.role)
result = await session.execute(stmt)
role_stats = result.all()

# Having clause
stmt = select(
    User.role,
    func.count(User.id).label('user_count')
).group_by(User.role).having(func.count(User.id) > 10)

# Distinct values
stmt = select(distinct(User.role))
result = await session.execute(stmt)
unique_roles = result.scalars().all()

Performance Optimization

The N+1 Problem

❌ Bad: N+1 Query Problem

# This will execute N+1 queries (1 + N for each user's posts)
async def get_users_with_posts_bad(session: AsyncSession) -> list[User]:
    stmt = select(User)
    result = await session.execute(stmt)
    users = result.scalars().all()
    
    for user in users:
        # This triggers a separate query for each user!
        posts = await user.awaitable_attrs.posts
    
    return users

✅ Good: Eager Loading Solutions

Option 1: selectinload (Recommended for One-to-Many)

async def get_users_with_posts_good(session: AsyncSession) -> list[User]:
    stmt = select(User).options(selectinload(User.posts))
    result = await session.execute(stmt)
    return result.scalars().all()

Option 2: joinedload (Good for One-to-One/Many-to-One)

async def get_posts_with_users(session: AsyncSession) -> list[Post]:
    stmt = select(Post).options(joinedload(Post.author))
    result = await session.execute(stmt)
    return result.scalars().all()

Option 3: Multiple relationships

async def get_users_with_full_data(session: AsyncSession) -> list[User]:
    stmt = select(User).options(
        selectinload(User.posts),
        selectinload(User.comments),
        joinedload(User.profile)
    )
    result = await session.execute(stmt)
    return result.scalars().all()

Lazy vs Eager Loading

# Lazy loading (default) - queries executed when accessed
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="author", lazy="select")

# Eager loading options
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    posts = relationship("Post", back_populates="author", lazy="selectin")  # Always eager load
    profile = relationship("Profile", back_populates="user", lazy="joined")  # Join load

Query Optimization Techniques

Selecting Only Required Columns

# Instead of loading full objects
stmt = select(User.id, User.email, User.username).where(User.is_active == True)
result = await session.execute(stmt)
user_data = result.all()

Using Indexes Effectively

# Make sure your model has proper indexes
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)  # Indexed for fast lookups
    created_at = Column(DateTime, index=True)        # Indexed for date queries
    is_active = Column(Boolean, index=True)          # Indexed for filtering
    
    # Composite index for common query patterns
    __table_args__ = (
        Index('idx_user_active_created', 'is_active', 'created_at'),
    )

Efficient Pagination

async def get_users_paginated(
    session: AsyncSession, 
    page: int = 1, 
    page_size: int = 20
) -> tuple[list[User], int]:
    """Get paginated users with total count."""
    
    # Count query
    count_stmt = select(func.count(User.id)).where(User.is_active == True)
    total_result = await session.execute(count_stmt)
    total = total_result.scalar()
    
    # Data query
    stmt = select(User).where(User.is_active == True).order_by(User.id).limit(page_size).offset((page - 1) * page_size)
    result = await session.execute(stmt)
    users = result.scalars().all()
    
    return users, total

Common Problems & Solutions

1. Avoiding SELECT * Queries

❌ Bad:

# Loads all columns unnecessarily
stmt = select(User)  # SELECT * FROM users

✅ Good:

# Load only needed columns
stmt = select(User.id, User.email, User.username)

2. Efficient Existence Checks

❌ Bad:

# Loads the entire object just to check existence
user = await session.get(User, user_id)
if user:
    # do something

✅ Good:

# Only checks existence
stmt = select(User.id).where(User.id == user_id)
result = await session.execute(stmt)
exists = result.first() is not None

3. Bulk Operations

❌ Bad:

# One query per record
for user_data in users_data:
    user = User(**user_data)
    session.add(user)
    await session.commit()  # Don't commit in loops!

✅ Good:

# Bulk insert
stmt = insert(User).values(users_data)
await session.execute(stmt)
await session.commit()  # Single commit

4. Complex Filtering

async def search_users(
    session: AsyncSession,
    search_term: str | None = None,
    role: str | None = None,
    is_active: bool | None = None,
    created_after: datetime | None = None
) -> list[User]:
    """Dynamic filtering based on provided parameters."""
    
    stmt = select(User)
    
    # Build conditions dynamically
    conditions = []
    
    if search_term:
        conditions.append(
            or_(
                User.username.ilike(f"%{search_term}%"),
                User.email.ilike(f"%{search_term}%")
            )
        )
    
    if role:
        conditions.append(User.role == role)
    
    if is_active is not None:
        conditions.append(User.is_active == is_active)
    
    if created_after:
        conditions.append(User.created_at >= created_after)
    
    if conditions:
        stmt = stmt.where(and_(*conditions))
    
    result = await session.execute(stmt)
    return result.scalars().all()

Advanced Techniques

Subqueries

# Subquery example: Users with more than 5 posts
subquery = select(Post.author_id).group_by(Post.author_id).having(func.count(Post.id) > 5).subquery()

stmt = select(User).where(User.id.in_(subquery))
result = await session.execute(stmt)
active_users = result.scalars().all()

Window Functions

from sqlalchemy import text

# Ranking users by score within their role
stmt = select(
    User.id,
    User.username,
    User.role,
    User.score,
    func.row_number().over(partition_by=User.role, order_by=User.score.desc()).label('rank')
).order_by(User.role, text('rank'))

result = await session.execute(stmt)
ranked_users = result.all()

Common Table Expressions (CTEs)

# Recursive CTE for hierarchical data
cte = select(User.id, User.name, User.manager_id, literal(0).label('level')).where(User.manager_id.is_(None)).cte(recursive=True)

cte_recursive = cte.union_all(
    select(
        User.id,
        User.name,
        User.manager_id,
        (cte.c.level + 1).label('level')
    ).join(cte, User.manager_id == cte.c.id)
)

stmt = select(cte_recursive)
result = await session.execute(stmt)
hierarchy = result.all()

Raw SQL for Complex Queries

async def complex_analytics_query(session: AsyncSession) -> list[dict]:
    """Use raw SQL for complex analytics that are hard to express in ORM."""
    
    query = text("""
        WITH monthly_stats AS (
            SELECT 
                DATE_TRUNC('month', created_at) as month,
                role,
                COUNT(*) as user_count,
                AVG(score) as avg_score
            FROM users 
            WHERE created_at >= :start_date
            GROUP BY DATE_TRUNC('month', created_at), role
        )
        SELECT 
            month,
            role,
            user_count,
            avg_score,
            LAG(user_count) OVER (PARTITION BY role ORDER BY month) as prev_month_count
        FROM monthly_stats
        ORDER BY month, role
    """)
    
    result = await session.execute(query, {"start_date": datetime.now() - timedelta(days=365)})
    return [dict(row._mapping) for row in result.all()]

Transaction Management

Basic Transactions

async def transfer_points(
    session: AsyncSession,
    from_user_id: int,
    to_user_id: int,
    amount: int
) -> bool:
    """Transfer points between users with transaction safety."""
    
    try:
        # Start transaction (auto-started with session)
        
        # Lock users for update to prevent race conditions
        from_user_stmt = select(User).where(User.id == from_user_id).with_for_update()
        to_user_stmt = select(User).where(User.id == to_user_id).with_for_update()
        
        from_user_result = await session.execute(from_user_stmt)
        to_user_result = await session.execute(to_user_stmt)
        
        from_user = from_user_result.scalar_one_or_none()
        to_user = to_user_result.scalar_one_or_none()
        
        if not from_user or not to_user:
            raise ValueError("User not found")
        
        if from_user.points < amount:
            raise ValueError("Insufficient points")
        
        # Update points
        from_user.points -= amount
        to_user.points += amount
        
        # Create transaction log
        transaction_log = TransactionLog(
            from_user_id=from_user_id,
            to_user_id=to_user_id,
            amount=amount,
            transaction_type="transfer"
        )
        session.add(transaction_log)
        
        await session.commit()
        return True
        
    except Exception as e:
        await session.rollback()
        raise e

Nested Transactions (Savepoints)

async def complex_operation_with_savepoints(session: AsyncSession):
    """Use savepoints for nested transaction control."""
    
    try:
        # Main operation
        user = User(email="user@example.com")
        session.add(user)
        
        # Savepoint for risky operation
        savepoint = await session.begin_nested()
        
        try:
            # Risky operation that might fail
            risky_operation()
            
        except Exception:
            # Rollback only to savepoint
            await savepoint.rollback()
            # Continue with main operation
        else:
            # Commit savepoint
            await savepoint.commit()
        
        # Commit main transaction
        await session.commit()
        
    except Exception:
        await session.rollback()
        raise

Best Practices

1. Repository Pattern Implementation

from abc import ABC, abstractmethod
from typing import Generic, TypeVar, Type

T = TypeVar('T')

class BaseRepository(ABC, Generic[T]):
    """Base repository with common database operations."""
    
    def __init__(self, session: AsyncSession, model: Type[T]):
        self.session = session
        self.model = model
    
    async def get_by_id(self, id: int) -> T | None:
        return await self.session.get(self.model, id)
    
    async def get_all(self, limit: int = 100, offset: int = 0) -> list[T]:
        stmt = select(self.model).limit(limit).offset(offset)
        result = await self.session.execute(stmt)
        return result.scalars().all()
    
    async def create(self, **kwargs) -> T:
        instance = self.model(**kwargs)
        self.session.add(instance)
        await self.session.commit()
        await self.session.refresh(instance)
        return instance
    
    async def update(self, id: int, **kwargs) -> T | None:
        instance = await self.get_by_id(id)
        if instance:
            for key, value in kwargs.items():
                setattr(instance, key, value)
            await self.session.commit()
            await self.session.refresh(instance)
        return instance
    
    async def delete(self, id: int) -> bool:
        instance = await self.get_by_id(id)
        if instance:
            await self.session.delete(instance)
            await self.session.commit()
            return True
        return False

class UserRepository(BaseRepository[User]):
    """User-specific repository methods."""
    
    async def get_by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_active_users(self) -> list[User]:
        stmt = select(User).where(User.is_active == True)
        result = await self.session.execute(stmt)
        return result.scalars().all()
    
    async def search_users(self, search_term: str) -> list[User]:
        stmt = select(User).where(
            or_(
                User.username.ilike(f"%{search_term}%"),
                User.email.ilike(f"%{search_term}%")
            )
        )
        result = await self.session.execute(stmt)
        return result.scalars().all()

2. Query Building Service

class QueryBuilderService:
    """Service for building complex dynamic queries."""
    
    @staticmethod
    def build_user_filter_query(
        base_query: Select,
        filters: dict
    ) -> Select:
        """Apply filters to user query."""
        
        if search := filters.get('search'):
            base_query = base_query.where(
                or_(
                    User.username.ilike(f"%{search}%"),
                    User.email.ilike(f"%{search}%")
                )
            )
        
        if role := filters.get('role'):
            base_query = base_query.where(User.role == role)
        
        if is_active := filters.get('is_active'):
            base_query = base_query.where(User.is_active == is_active)
        
        if created_after := filters.get('created_after'):
            base_query = base_query.where(User.created_at >= created_after)
        
        if created_before := filters.get('created_before'):
            base_query = base_query.where(User.created_at <= created_before)
        
        # Sorting
        if sort_by := filters.get('sort_by'):
            direction = filters.get('sort_direction', 'asc')
            column = getattr(User, sort_by, None)
            if column:
                base_query = base_query.order_by(
                    column.desc() if direction == 'desc' else column.asc()
                )
        
        return base_query

3. Performance Monitoring

import time
from functools import wraps

def log_query_performance(func):
    """Decorator to log query execution time."""
    
    @wraps(func)
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await func(*args, **kwargs)
        execution_time = time.time() - start_time
        
        if execution_time > 1.0:  # Log slow queries
            logger.warning(
                f"Slow query detected: {func.__name__} took {execution_time:.2f}s"
            )
        
        return result
    
    return wrapper

# Usage
@log_query_performance
async def get_users_with_complex_data(session: AsyncSession) -> list[User]:
    # Complex query implementation
    pass

Examples

Complete CRUD Service

class UserService:
    """Complete user service with optimized queries."""
    
    def __init__(self, session: AsyncSession):
        self.session = session
    
    async def create_user(self, user_data: dict) -> User:
        """Create a new user."""
        user = User(**user_data)
        self.session.add(user)
        await self.session.commit()
        await self.session.refresh(user)
        return user
    
    async def get_user_with_posts(self, user_id: int) -> User | None:
        """Get user with their posts (optimized)."""
        stmt = select(User).options(
            selectinload(User.posts)
        ).where(User.id == user_id)
        
        result = await self.session.execute(stmt)
        return result.scalar_one_or_none()
    
    async def get_users_paginated(
        self,
        page: int = 1,
        page_size: int = 20,
        filters: dict = None
    ) -> tuple[list[User], int]:
        """Get paginated users with filters."""
        
        base_query = select(User)
        count_query = select(func.count(User.id))
        
        if filters:
            base_query = QueryBuilderService.build_user_filter_query(base_query, filters)
            count_query = QueryBuilderService.build_user_filter_query(count_query, filters)
        
        # Get total count
        total_result = await self.session.execute(count_query)
        total = total_result.scalar()
        
        # Get paginated data
        stmt = base_query.limit(page_size).offset((page - 1) * page_size)
        result = await self.session.execute(stmt)
        users = result.scalars().all()
        
        return users, total
    
    async def update_user_last_login(self, user_id: int) -> None:
        """Update user's last login time (optimized)."""
        stmt = update(User).where(User.id == user_id).values(
            last_login=datetime.now()
        )
        await self.session.execute(stmt)
        await self.session.commit()
    
    async def get_user_statistics(self) -> dict:
        """Get user statistics with aggregations."""
        stmt = select(
            func.count(User.id).label('total_users'),
            func.count(User.id).filter(User.is_active == True).label('active_users'),
            func.count(distinct(User.role)).label('unique_roles'),
            func.avg(User.score).label('average_score')
        )
        
        result = await self.session.execute(stmt)
        row = result.first()
        
        return {
            'total_users': row.total_users,
            'active_users': row.active_users,
            'unique_roles': row.unique_roles,
            'average_score': float(row.average_score) if row.average_score else 0
        }

Performance Tips Summary

1. Always use eager loading for relationships you know you'll need
2. Select only required columns when you don't need full objects
3. Use bulk operations for multiple records
4. Add proper indexes to your models
5. Monitor query performance and log slow queries
6. Use connection pooling (already configured in the template)
7. Implement query caching for frequently accessed data
8. Use raw SQL for complex analytics queries
9. Batch database operations when possible
10. Always use transactions for data consistency

---

See Also:

• Database Models - Creating and managing SQLAlchemy models
• Migrations Guide - Managing database schema changes
• Testing Guide - Testing database operations
• Performance Optimization - Advanced performance techniques