14.5 SQLAlchemy Complex Queries

NicePython

Oct 4, 2024 — 5 min read

While basic CRUD (Create, Read, Update, Delete) operations are essential for most applications, complex queries are often required to handle more sophisticated data retrieval and manipulation. SQLAlchemy provides robust support for writing complex SQL queries, making it easier to interact with the database using Pythonic code.

In this section, we’ll explore how to write and execute complex queries using SQLAlchemy's ORM and Core APIs. We'll cover:

Joins
Filtering with complex conditions
Aggregations and groupings
Subqueries
Using raw SQL when needed

14.5.1 Joins in SQLAlchemy

A join combines rows from two or more tables based on a related column between them. SQLAlchemy allows you to perform inner joins, left joins, and outer joins easily using its ORM.

Example: Defining Models for Joining

Assume we have two models: User and Address. Each user can have multiple addresses, and each address is linked to a specific user.

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from .database import Base

# User model
class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    email = Column(String, unique=True)
    
    # Relationship with Address
    addresses = relationship("Address", back_populates="user")

# Address model
class Address(Base):
    __tablename__ = 'addresses'

    id = Column(Integer, primary_key=True, index=True)
    street = Column(String)
    city = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))

    # Relationship with User
    user = relationship("User", back_populates="addresses")

User: Has a one-to-many relationship with the Address model.
Address: Contains a foreign key (user_id) linking it to the User model.

1. Performing an Inner Join

To retrieve users and their addresses, you can use an inner join. An inner join only returns rows that have matching records in both tables.

from sqlalchemy.orm import Session
from sqlalchemy import select
from .models import User, Address

# Inner join between User and Address
def get_users_with_addresses(db: Session):
    stmt = select(User, Address).join(Address, User.id == Address.user_id)
    result = db.execute(stmt).all()
    
    for user, address in result:
        print(f"User: {user.name}, Address: {address.street}, {address.city}")

This query:

Joins the User table with the Address table on the user_id column.
Returns each user with their associated addresses.

2. Performing a Left Join

A left join returns all rows from the left table (in this case, User) and matching rows from the right table (Address). If there are no matches, it returns NULL values for the columns from the right table.

# Left join between User and Address
def get_users_with_or_without_addresses(db: Session):
    stmt = select(User, Address).outerjoin(Address, User.id == Address.user_id)
    result = db.execute(stmt).all()
    
    for user, address in result:
        if address:
            print(f"User: {user.name}, Address: {address.street}, {address.city}")
        else:
            print(f"User: {user.name}, Address: None")

This query returns all users, even those who do not have associated addresses.

14.5.2 Filtering and Complex Conditions

SQLAlchemy provides several ways to filter data with complex conditions using the filter() and filter_by() methods.

1. Simple Filtering

You can filter records using conditions like equal to, greater than, like, etc.

# Get all users older than 30
def get_users_older_than_30(db: Session):
    users = db.query(User).filter(User.age > 30).all()
    for user in users:
        print(f"User: {user.name}, Age: {user.age}")

2. Using Multiple Conditions

You can combine multiple conditions using AND (&) and OR (|) operators.

from sqlalchemy import and_, or_

# Get users who are either older than 30 or live in "New York"
def get_filtered_users(db: Session):
    users = db.query(User).filter(or_(User.age > 30, User.city == "New York")).all()
    for user in users:
        print(f"User: {user.name}, Age: {user.age}, City: {user.city}")

In this query, we filter users who are older than 30 or live in New York using the or_() function.

14.5.3 Aggregations and Grouping

SQLAlchemy provides support for SQL aggregation functions such as COUNT(), SUM(), AVG(), and GROUP BY.

1. Counting Records

The count() function can be used to count the number of records in a table.

# Count the number of users in the database
def get_user_count(db: Session):
    count = db.query(User).count()
    print(f"Total Users: {count}")

2. Grouping Data

You can group data using the group_by() method and apply aggregate functions like COUNT() or SUM() to each group.

from sqlalchemy import func

# Get the count of addresses per user
def get_address_count_per_user(db: Session):
    address_count = db.query(User.name, func.count(Address.id)).join(Address).group_by(User.id).all()
    
    for name, count in address_count:
        print(f"User: {name}, Address Count: {count}")

In this query:

We group by User.id and use func.count() to count the number of addresses associated with each user.

14.5.4 Subqueries

SQLAlchemy supports subqueries, which allow you to embed one query within another. This is useful when you need to perform more complex queries that rely on intermediate results.

Example: Subquery for Users with Multiple Addresses

Let’s say we want to find users who have more than one address.

# Get users with more than one address using a subquery
def get_users_with_multiple_addresses(db: Session):
    subquery = db.query(Address.user_id, func.count(Address.id).label('address_count')).group_by(Address.user_id).subquery()

    users = db.query(User).join(subquery, User.id == subquery.c.user_id).filter(subquery.c.address_count > 1).all()
    
    for user in users:
        print(f"User: {user.name}, has multiple addresses")

In this example:

We use a subquery to count the number of addresses per user and filter for users with more than one address.

14.5.5 Using Raw SQL Queries

Although SQLAlchemy provides an abstraction over SQL, there might be times when you need to execute raw SQL for performance reasons or for complex queries not easily expressible in the ORM.

Example: Executing Raw SQL

You can execute raw SQL using the execute() method on the session or engine object.

# Execute raw SQL
def run_raw_sql(db: Session):
    result = db.execute("SELECT * FROM users WHERE age > 30")
    for row in result:
        print(row)

This query bypasses the ORM and runs a raw SQL query directly on the database.

14.5.6 Combining ORM and SQL Expression Language

SQLAlchemy allows you to combine the ORM with SQL expressions for more complex queries. For instance, you can mix ORM queries with Core expressions, aggregations, and filters.

Example: Using SQL Functions in ORM Queries

You can use SQLAlchemy’s built-in SQL functions like func to integrate SQL functions with ORM queries.

from sqlalchemy import func

# Get users whose email domains end with 'example.com'
def get_users_by_email_domain(db: Session):
    users = db.query(User).filter(func.substring(User.email, -11) == 'example.com').all()
    
    for user in users:
        print(f"User: {user.name}, Email: {user.email}")

In this query:

We use the func.substring() function to filter users based on their email domain.

14.5.7 Complex Queries with Aliases

Aliases are useful for self-joins or when you need to refer to the same table multiple times in a query.

Example: Self-Join with Aliases

Let’s assume we want to find users who share the same city with other users.

from sqlalchemy.orm import aliased

# Find users who share the same city
def get_users_in_same_city(db: Session):
    user_alias = aliased(User)

    result = db.query(User.name, user_alias.name).filter(User.city == user_alias.city, User.id != user_alias.id).all()
    
    for user1, user2 in result:
        print(f"{user1} and {user2} live in the same

 city")