← Python

Database & ORM (SQLAlchemy, raw SQL, migrations)

← Database & ORM (SQLAlchemy, raw SQL, migrations)

Select a question from the sidebar

15 questions in this topic

Question 1 / 15

What is SQLAlchemy and how does it differ from raw SQL?

SQLAlchemy is a Python SQL toolkit and ORM that provides two main layers:

SQLAlchemy Core - A SQL abstraction layer that lets you write SQL using Python expressions. It provides a Table, Column, and select() API that compiles to SQL for any supported database.
SQLAlchemy ORM - A full Object-Relational Mapper built on top of Core. It maps Python classes to database tables and Python objects to rows, allowing you to work with data as native Python objects instead of raw tuples.

Key differences from raw SQL:

Portability: Same Python code works across PostgreSQL, MySQL, SQLite, etc.
Security: Parameterized queries by default prevent SQL injection.
Abstraction: Work with Python objects instead of result tuples.
Relationship management: Automatic JOIN generation through relationship definitions.
Session/Unit of Work: Changes are batched and committed together, providing transactional integrity.

Pseudo Code

1. Import create_engine and declarative_base
2. Define engine with connection URL
3. Create a Base class for models
4. Define model classes that inherit from Base
5. Create tables with Base.metadata.create_all(engine)
6. Create a Session to interact with the database
7. Use session.add(), session.query(), session.commit()

Code

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

# Core: create engine
engine = create_engine('sqlite:///example.db', echo=True)

# ORM: declarative base
Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(120), unique=True)

    def __repr__(self):
        return f"<User(name='{self.name}', email='{self.email}')>"

# Create tables
Base.metadata.create_all(engine)

# Use the ORM
with Session(engine) as session:
    user = User(name='Alice', email='alice@example.com')
    session.add(user)
    session.commit()

    # Query
    found = session.query(User).filter_by(name='Alice').first()
    print(found)  # <User(name='Alice', email='alice@example.com')>

Key Points

•SQLAlchemy has two layers: Core (SQL expression language) and ORM (object mapping)
•The engine manages database connections and connection pooling
•declarative_base() creates a base class that links Python classes to database tables
•The Session is the primary interface for persistence operations in ORM
•SQLAlchemy automatically parameterizes queries to prevent SQL injection

Gotchas

•echo=True in create_engine logs all SQL, which is useful for debugging but noisy in production
•Forgetting session.commit() means your changes are never written to the database
•SQLAlchemy 2.0 uses declarative_base from sqlalchemy.orm, not sqlalchemy.ext.declarative

SQLAlchemy is a Python ORM and SQL toolkit. Its ORM maps Python classes to database tables, letting you query and manipulate data as Python objects instead of writing raw SQL. It also provides connection pooling, portability across databases, and automatic SQL injection prevention.

•Two layers: Core (SQL expressions) and ORM (object mapping)
•Portable across databases, parameterized by default

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))

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

with Session(engine) as session:
    session.add(User(name='Alice'))
    session.commit()

Always call session.commit() or your changes are lost

Question 2 / 15

How do you define a SQLAlchemy declarative model with columns and constraints?

A declarative model in SQLAlchemy is a Python class that inherits from a declarative base and maps to a database table. You define columns using Column() objects and add constraints either inline or as table-level arguments.

Column types: Integer, String(length), Float, Boolean, DateTime, Text, Numeric, LargeBinary, Enum.

Column options:

primary_key=True - Makes the column a primary key
nullable=False - NOT NULL constraint
unique=True - UNIQUE constraint
default=value - Default value (Python-side)
server_default='value' - Default value (database-side)
index=True - Creates an index on the column

Table-level constraints: Use __table_args__ for composite constraints like unique together, composite indexes, or check constraints.

Pseudo Code

1. Import Column types from sqlalchemy
2. Create a class inheriting from Base
3. Set __tablename__ to the desired table name
4. Define columns with Column() and appropriate types
5. Add constraints inline or via __table_args__
6. Optionally add __repr__ for debugging

Code

from sqlalchemy import (
    Column, Integer, String, Float, Boolean,
    DateTime, Text, ForeignKey, UniqueConstraint,
    CheckConstraint, Index
)
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'

    # Primary key (auto-increment by default for Integer)
    id = Column(Integer, primary_key=True)

    # Required string with max length
    name = Column(String(200), nullable=False)

    # Optional text field
    description = Column(Text, nullable=True)

    # Numeric with constraints
    price = Column(Float, nullable=False)
    quantity = Column(Integer, default=0)

    # Boolean with server-side default
    is_active = Column(Boolean, server_default='1')

    # DateTime with Python-side default
    created_at = Column(DateTime, default=datetime.utcnow)

    # Indexed and unique columns
    sku = Column(String(50), unique=True, index=True)

    # Foreign key
    category_id = Column(Integer, ForeignKey('categories.id'), nullable=False)

    # Table-level constraints
    __table_args__ = (
        UniqueConstraint('name', 'category_id', name='uq_product_name_category'),
        CheckConstraint('price >= 0', name='ck_positive_price'),
        Index('ix_product_category_active', 'category_id', 'is_active'),
    )

    def __repr__(self):
        return f"<Product(name='{self.name}', price={self.price})>"

Key Points

•__tablename__ is required and maps the class to a database table
•Column types map to SQL types: String -> VARCHAR, Integer -> INTEGER, Text -> TEXT
•default= runs in Python, server_default= adds a DEFAULT in the SQL schema
•__table_args__ is a tuple of table-level constraints and must end with a comma if single element
•ForeignKey references another table's column as 'tablename.column'

Gotchas

•String requires a length argument for most databases (except SQLite), e.g., String(50)
•default=datetime.utcnow (no parentheses) passes the function; default=datetime.utcnow() passes a fixed timestamp
•__table_args__ must be a tuple, not a list; for a single constraint use a trailing comma: (constraint,)

Define a class inheriting from declarative_base(), set __tablename__, and add Column() attributes with types like Integer, String, Float, etc. Constraints go inline (nullable=False, unique=True) or in __table_args__ for composite constraints.

•__tablename__ maps the class to a table; columns use Column() with SQL types
•Use __table_args__ tuple for composite constraints

from sqlalchemy import Column, Integer, String, UniqueConstraint
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(120), unique=True, index=True)

    __table_args__ = (
        UniqueConstraint('name', 'email', name='uq_user_name_email'),
    )

default=func passes the callable; default=func() passes a fixed value computed once

Question 3 / 15

How do you define one-to-many relationships in SQLAlchemy?

A one-to-many relationship means one record in a parent table relates to multiple records in a child table. In SQLAlchemy, this is defined with:

A ForeignKey column on the child (many side) pointing to the parent's primary key.
A relationship() on the parent (one side) that provides a Python-level collection of children.
Optionally, a back_populates or backref to create the reverse reference.

back_populates vs backref:

back_populates requires defining relationship() on both sides explicitly (recommended for clarity).
backref auto-creates the reverse relationship on the other model (less code but implicit).

Cascade options: Control what happens to children when parent operations occur. Common cascades: all, delete-orphan, save-update, merge.

Pseudo Code

1. Define parent model with a relationship() attribute
2. Define child model with a ForeignKey column to parent
3. Add back_populates on both sides (or backref on one side)
4. Configure cascade behavior if needed
5. Access children via parent.children or parent via child.parent

Code

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

Base = declarative_base()

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)

    # One author has many books
    books = relationship('Book', back_populates='author',
                         cascade='all, delete-orphan')

    def __repr__(self):
        return f"<Author(name='{self.name}')>"

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    author_id = Column(Integer, ForeignKey('authors.id'), nullable=False)

    # Each book belongs to one author
    author = relationship('Author', back_populates='books')

    def __repr__(self):
        return f"<Book(title='{self.title}')>"

# Usage
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    author = Author(name='George Orwell')
    author.books.append(Book(title='1984'))
    author.books.append(Book(title='Animal Farm'))
    session.add(author)
    session.commit()

    # Access the relationship
    loaded = session.query(Author).first()
    print(loaded.books)  # [<Book(title='1984')>, <Book(title='Animal Farm')>]
    print(loaded.books[0].author)  # <Author(name='George Orwell')>

[<Book(title='1984')>, <Book(title='Animal Farm')>]
<Author(name='George Orwell')>

Key Points

•ForeignKey goes on the child (many side); relationship() goes on the parent (one side) or both
•back_populates requires explicit relationship() on both models; backref auto-creates it
•cascade='all, delete-orphan' deletes children when the parent is deleted
•You can append to the collection (author.books.append) and SQLAlchemy handles the ForeignKey

Gotchas

•Forgetting nullable=False on ForeignKey allows orphan records with NULL parent references
•Without cascade='delete-orphan', removing a book from author.books leaves it in the DB with a NULL FK
•back_populates values must match the attribute names exactly on each side

Put a ForeignKey on the child model pointing to the parent, then use relationship() on the parent with back_populates to link both directions. Optionally set cascade='all, delete-orphan' to auto-delete children when the parent is removed.

•ForeignKey on child, relationship() on parent (or both with back_populates)
•cascade='all, delete-orphan' for automatic child cleanup

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    books = relationship('Book', back_populates='author',
                         cascade='all, delete-orphan')

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    author_id = Column(Integer, ForeignKey('authors.id'))
    author = relationship('Author', back_populates='books')

back_populates strings must exactly match the attribute name on the other model

Question 4 / 15

How do you define many-to-many relationships in SQLAlchemy?

Many-to-many relationships require an association table (also called a junction/bridge table) that holds foreign keys to both sides. SQLAlchemy offers two approaches:

Simple association table using Table() - When the join table only has the two foreign keys and no extra columns.
Association object using a full model class - When the join table has additional columns (e.g., enrolled_date, grade).

For the simple case, define the association table with Table(), then use relationship() with the secondary parameter pointing to that table.

Pseudo Code

1. Define an association table with Table() containing two ForeignKeys
2. On one model, define relationship() with secondary=association_table
3. On the other model, define relationship() with back_populates
4. Use collection.append() to add links
5. For extra columns, use an Association Object pattern instead

Code

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import declarative_base, relationship, Session

Base = declarative_base()

# Simple association table (no extra columns)
student_course = Table(
    'student_course', Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True),
)

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    courses = relationship('Course', secondary=student_course,
                           back_populates='students')

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    students = relationship('Student', secondary=student_course,
                            back_populates='courses')

# Usage
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    math = Course(title='Mathematics')
    physics = Course(title='Physics')
    alice = Student(name='Alice', courses=[math, physics])
    bob = Student(name='Bob', courses=[math])

    session.add_all([alice, bob])
    session.commit()

    # Query through the relationship
    math_students = session.query(Course).filter_by(title='Mathematics').first()
    print([s.name for s in math_students.students])  # ['Alice', 'Bob']

# --- Association Object (with extra columns) ---
class Enrollment(Base):
    __tablename__ = 'enrollments'
    student_id = Column(Integer, ForeignKey('students.id'), primary_key=True)
    course_id = Column(Integer, ForeignKey('courses.id'), primary_key=True)
    grade = Column(String(2))
    student = relationship('Student', backref='enrollments')
    course = relationship('Course', backref='enrollments')

['Alice', 'Bob']

Key Points

•Use Table() for simple join tables with only foreign keys
•Pass secondary=table to relationship() on both sides
•Use an Association Object (full model class) when the join table has extra columns
•Both columns in the association table should be primary keys (composite PK)

Gotchas

•Using secondary= with an Association Object model causes conflicts; use explicit relationships instead
•Deleting from the collection (student.courses.remove(course)) only deletes the association row, not the course
•The association Table must be defined before the models that reference it, or use a string reference

Create an association Table() with two ForeignKey columns, then use relationship(secondary=table) on both models. For extra columns on the join table, use an Association Object (a full model class) instead.

•Table() with two ForeignKey PKs + relationship(secondary=table) on both sides
•Use Association Object pattern when join table needs extra columns

student_course = Table(
    'student_course', Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True),
)

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    courses = relationship('Course', secondary=student_course, back_populates='students')

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    title = Column(String(200))
    students = relationship('Student', secondary=student_course, back_populates='courses')

Do not use secondary= alongside an Association Object model

Question 5 / 15

How do you query with filter, filter_by, and common query methods in SQLAlchemy?

SQLAlchemy provides a rich Query API for retrieving data. The two main filtering methods are:

**filter_by(**kwargs)** - Simple equality filters using keyword arguments. Good for simple column = value checks.
filter(expression) - Full SQL expression filters. Supports all operators: ==, !=, <, >, like(), in_(), between(), is_(None), and_(), or_().

Result methods:

.all() - Returns a list of all results
.first() - Returns the first result or None
.one() - Returns exactly one result; raises if 0 or >1
.one_or_none() - Returns one result or None; raises if >1
.scalar() - Returns the first column of the first row
.count() - Returns the count of rows

Ordering and limiting:

.order_by(Model.column.desc())
.limit(n) and .offset(n)
.distinct()

Pseudo Code

1. session.query(Model) starts a query
2. Chain .filter() or .filter_by() for conditions
3. Use .order_by(), .limit(), .offset() as needed
4. Terminate with .all(), .first(), .one(), or .count()
5. Use and_(), or_() from sqlalchemy for complex conditions

Code

from sqlalchemy import and_, or_, func
from sqlalchemy.orm import Session

with Session(engine) as session:
    # filter_by: keyword equality only
    user = session.query(User).filter_by(name='Alice', is_active=True).first()

    # filter: full expression support
    users = session.query(User).filter(User.age >= 18).all()

    # Comparison operators
    session.query(User).filter(User.name != 'Bob').all()
    session.query(User).filter(User.age.between(18, 30)).all()
    session.query(User).filter(User.name.like('%ali%')).all()
    session.query(User).filter(User.name.ilike('%ALI%')).all()  # case-insensitive
    session.query(User).filter(User.id.in_([1, 2, 3])).all()
    session.query(User).filter(User.email.is_(None)).all()

    # Combining conditions
    session.query(User).filter(
        and_(User.age >= 18, User.is_active == True)
    ).all()

    session.query(User).filter(
        or_(User.name == 'Alice', User.name == 'Bob')
    ).all()

    # Chaining filters (implicit AND)
    session.query(User).filter(User.age >= 18).filter(User.is_active == True).all()

    # Ordering and limiting
    session.query(User).order_by(User.name.asc()).limit(10).offset(20).all()

    # Aggregation
    count = session.query(func.count(User.id)).scalar()
    avg_age = session.query(func.avg(User.age)).scalar()

    # Result terminators
    first_user = session.query(User).first()         # first or None
    exact_one = session.query(User).filter_by(id=1).one()  # exactly 1 or raise
    maybe_one = session.query(User).filter_by(id=1).one_or_none()

Key Points

•filter_by() uses keyword args (equality only); filter() uses SQL expressions (full operators)
•Chaining .filter() multiple times is an implicit AND
•.first() returns None if no match; .one() raises NoResultFound or MultipleResultsFound
•Use func.count(), func.avg(), func.sum() for aggregations
•.in_() takes a list; .like() and .ilike() for pattern matching

Gotchas

•Comparing to None with == None works but .is_(None) is more explicit and correct SQL
•.one() raises MultipleResultsFound if the query returns more than one row
•filter_by uses Python keyword syntax (name='x'), filter uses SQL expression syntax (Model.name == 'x')

filter_by(col=val) is for simple equality; filter(Model.col == val) supports all SQL operators. Chain .all(), .first(), .one(), or .count() to get results. Chain multiple .filter() calls for implicit AND, or use and_() / or_() for complex logic.

•filter_by for equality kwargs; filter for full SQL expressions
•.first() returns None; .one() raises on 0 or >1 results

# Simple equality
user = session.query(User).filter_by(name='Alice').first()

# Complex expression
users = session.query(User).filter(
    and_(User.age >= 18, User.is_active == True)
).order_by(User.name).limit(10).all()

# Aggregation
count = session.query(func.count(User.id)).scalar()

Use .is_(None) instead of == None for proper SQL IS NULL

Question 6 / 15

How do you perform JOINs and subqueries in SQLAlchemy ORM?

SQLAlchemy ORM supports several JOIN approaches:

Implicit joins via relationships: When you access a relationship attribute, SQLAlchemy generates the JOIN.
Explicit .join(): Chain .join() on a query to perform inner joins.
.outerjoin(): For LEFT OUTER JOINs.
Subqueries: Use .subquery() to wrap a query as a derived table, then join or filter against it.
contains_eager(): Tell SQLAlchemy you are already joining a relationship, so it can populate the attribute from the joined data.

The .join() method can take a model class (auto-detects the FK), an explicit ON condition, or a relationship attribute.

Pseudo Code

1. Basic join: session.query(A).join(B).filter(B.col == val)
2. Outer join: session.query(A).outerjoin(B)
3. Subquery: sub = session.query(...).subquery()
4. Use sub.c.column_name to reference subquery columns
5. Join against subquery: session.query(A).join(sub, A.id == sub.c.a_id)

Code

from sqlalchemy import func
from sqlalchemy.orm import Session, contains_eager

with Session(engine) as session:
    # Basic inner join (auto-detect FK)
    results = (
        session.query(Author, Book)
        .join(Book)
        .filter(Book.title.like('%Python%'))
        .all()
    )
    # Each result is a tuple of (Author, Book)

    # Join with explicit condition
    results = (
        session.query(Author)
        .join(Book, Author.id == Book.author_id)
        .filter(Book.price > 20)
        .all()
    )

    # Left outer join
    results = (
        session.query(Author, func.count(Book.id).label('book_count'))
        .outerjoin(Book)
        .group_by(Author.id)
        .all()
    )

    # Subquery: find authors with more than 5 books
    subq = (
        session.query(
            Book.author_id,
            func.count(Book.id).label('book_count')
        )
        .group_by(Book.author_id)
        .having(func.count(Book.id) > 5)
        .subquery()
    )

    prolific_authors = (
        session.query(Author)
        .join(subq, Author.id == subq.c.author_id)
        .all()
    )

    # contains_eager: populate relationship from explicit join
    authors_with_books = (
        session.query(Author)
        .join(Author.books)
        .options(contains_eager(Author.books))
        .filter(Book.is_published == True)
        .all()
    )

Key Points

•.join(Model) auto-detects the foreign key; .join(Model, condition) for explicit ON clause
•.outerjoin() produces LEFT OUTER JOIN
•.subquery() wraps a query as a derived table; access columns with .c.column_name
•contains_eager() populates a relationship from an already-joined table
•Use .label('name') to give computed columns a reference name

Gotchas

•Joining without relationships requires an explicit ON condition if there are multiple FKs between tables
•subquery().c gives you column references, not model attributes
•A .join() without .filter() on the joined table still restricts to rows with matches (inner join behavior)

Use .join(Model) for INNER JOINs, .outerjoin(Model) for LEFT JOINs. For subqueries, create a query, call .subquery(), then join against it using subq.c.column to reference its columns.

•.join() for inner join, .outerjoin() for left join
•.subquery() + .c.column for derived table references

# Inner join
results = session.query(Author).join(Book).filter(Book.price > 20).all()

# Subquery
subq = session.query(Book.author_id, func.count(Book.id).label('cnt')
    ).group_by(Book.author_id).having(func.count(Book.id) > 5).subquery()

authors = session.query(Author).join(subq, Author.id == subq.c.author_id).all()

Subquery columns are accessed via .c, not as model attributes

Question 7 / 15

What is the N+1 query problem and how do you solve it in SQLAlchemy?

The N+1 problem occurs when you load N parent objects and then access a relationship on each one, triggering a separate SQL query for each parent. So 1 query loads the parents, then N additional queries load the children -- hence N+1 total queries.

This is caused by lazy loading, which is SQLAlchemy's default: relationships are not loaded until you access them.

Solutions (eager loading strategies):

joinedload() - Uses a LEFT OUTER JOIN to load the relationship in the same query. Best for one-to-one and many-to-one.
subqueryload() - Fires a second query using a subquery (SELECT ... WHERE parent_id IN (SELECT ...)). Best for one-to-many collections.
selectinload() - Fires a second query using IN: SELECT ... WHERE parent_id IN (1,2,3...). Generally the best default for collections.
lazy='joined' on the relationship definition - Always eager load.
contains_eager() - Use when you have already joined manually.

Pseudo Code

1. Identify the N+1: accessing relationship in a loop triggers lazy loads
2. Add .options(selectinload(Parent.children)) to the query
3. Or use joinedload for to-one relationships
4. Or set lazy='selectin' on the relationship definition
5. Verify with echo=True that only 1-2 queries are fired

Code

from sqlalchemy.orm import selectinload, joinedload, subqueryload, Session

# THE PROBLEM: N+1 queries
with Session(engine) as session:
    authors = session.query(Author).all()  # Query 1
    for author in authors:
        print(author.books)  # Query 2, 3, 4, ... N+1 (one per author!)

# SOLUTION 1: selectinload (best general-purpose for collections)
with Session(engine) as session:
    authors = (
        session.query(Author)
        .options(selectinload(Author.books))
        .all()
    )
    # Only 2 queries total:
    # 1) SELECT * FROM authors
    # 2) SELECT * FROM books WHERE author_id IN (1, 2, 3, ...)
    for author in authors:
        print(author.books)  # No additional queries

# SOLUTION 2: joinedload (best for to-one or small collections)
with Session(engine) as session:
    books = (
        session.query(Book)
        .options(joinedload(Book.author))
        .all()
    )
    # 1 query with LEFT OUTER JOIN
    for book in books:
        print(book.author)  # Already loaded

# SOLUTION 3: subqueryload
with Session(engine) as session:
    authors = (
        session.query(Author)
        .options(subqueryload(Author.books))
        .all()
    )

# SOLUTION 4: Set default on the relationship
class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    books = relationship('Book', back_populates='author',
                         lazy='selectin')  # Always eager load

# Nested eager loading
with Session(engine) as session:
    authors = (
        session.query(Author)
        .options(
            selectinload(Author.books)
            .selectinload(Book.reviews)  # Also eager load reviews of books
        )
        .all()
    )

Key Points

•N+1 = 1 query for parents + N queries for children when accessing lazy-loaded relationships in a loop
•selectinload() is the best general-purpose solution for collections (uses IN clause)
•joinedload() uses a JOIN and is best for to-one relationships
•You can chain eager loads for nested relationships: selectinload(A.b).selectinload(B.c)
•Set lazy='selectin' on relationship() to always eager load

Gotchas

•joinedload on large collections can produce massive result sets due to the Cartesian product
•joinedload may interfere with .limit() and .offset() because the JOIN multiplies rows
•Setting echo=True on the engine is the quickest way to detect N+1 problems during development

N+1 means loading N parents triggers N extra queries to load their children (lazy loading). Fix it with eager loading: selectinload() (uses IN clause, best for collections), joinedload() (uses JOIN, best for to-one), or set lazy='selectin' on the relationship.

•selectinload for collections, joinedload for to-one relationships
•Lazy loading is the default and causes N+1 when accessed in loops

from sqlalchemy.orm import selectinload

# Bad: N+1
authors = session.query(Author).all()
for a in authors:
    print(a.books)  # triggers a query per author

# Good: 2 queries total
authors = session.query(Author).options(selectinload(Author.books)).all()
for a in authors:
    print(a.books)  # already loaded

joinedload with .limit() can return fewer results than expected due to JOIN row multiplication

Question 8 / 15

How do eager loading and lazy loading differ in SQLAlchemy?

Lazy loading (default): Related objects are not loaded from the database until you access the relationship attribute. Each access triggers a new SQL query.

Eager loading: Related objects are loaded upfront, either in the same query or in a follow-up query, before you access them.

SQLAlchemy offers several loading strategies set via lazy= parameter on relationship() or via query .options():

Strategy	`lazy=`	`.options()`	SQL Behavior
Lazy (select)	`'select'` (default)	N/A	Separate SELECT on access
Joined	`'joined'`	`joinedload()`	LEFT OUTER JOIN
Subquery	`'subquery'`	`subqueryload()`	Subquery IN clause
Select In	`'selectin'`	`selectinload()`	SELECT ... WHERE id IN (...)
Raise	`'raise'`	`raiseload()`	Raises error if accessed (prevents lazy)
No load	`'noload'`	`noload()`	Never loads, returns default (None/[])

Pseudo Code

1. Default is lazy='select' - loads on attribute access
2. Use .options(joinedload/selectinload/subqueryload) per-query
3. Or set lazy='selectin' on the relationship for global behavior
4. Use lazy='raise' to catch accidental lazy loads during development
5. raiseload('*') prevents lazy loading on all relationships of a model

Code

from sqlalchemy.orm import (
    relationship, Session,
    joinedload, selectinload, subqueryload, raiseload, noload
)

# Strategy on the relationship definition
class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))

    # Always eager load with selectin
    books = relationship('Book', lazy='selectin', back_populates='author')

# Strategy per-query (overrides relationship setting)
with Session(engine) as session:
    # joinedload: single query with JOIN
    authors = session.query(Author).options(joinedload(Author.books)).all()

    # selectinload: 2 queries (parent + IN clause)
    authors = session.query(Author).options(selectinload(Author.books)).all()

    # raiseload: raises InvalidRequestError if you try to lazy load
    authors = session.query(Author).options(raiseload(Author.books)).all()
    # authors[0].books  # Raises! Forces you to eager load explicitly.

    # Wildcard raiseload: prevent ALL lazy loading
    authors = session.query(Author).options(raiseload('*')).all()

    # noload: relationship is never loaded (returns None or empty list)
    authors = session.query(Author).options(noload(Author.books)).all()
    print(authors[0].books)  # [] (empty, not loaded)

Key Points

•Lazy (select) is the default: loads on attribute access with a new query
•joinedload uses a JOIN, selectinload uses an IN clause, subqueryload uses a subquery
•raiseload prevents accidental lazy loading by raising an error
•Per-query .options() override the relationship-level lazy= setting
•selectinload is generally the safest default for collections

Gotchas

•Lazy loading works only within an active session; accessing outside raises DetachedInstanceError
•joinedload can produce duplicated parent rows that SQLAlchemy de-duplicates but it inflates data transfer
•raiseload('*') is a great development tool but make sure you don't deploy it without proper eager loading

Lazy loading (default) loads related objects on access, triggering a new query each time. Eager loading loads them upfront via JOIN (joinedload), IN clause (selectinload), or subquery (subqueryload). Set strategy on the relationship (lazy='selectin') or per-query (.options(selectinload(...))).

•Default is lazy='select'; best default eager strategy is selectinload
•Use raiseload to catch unintended lazy loads

# Lazy (default) - new query on access
class Author(Base):
    books = relationship('Book', lazy='select')  # default

# Eager - set globally
class Author(Base):
    books = relationship('Book', lazy='selectin')

# Eager - per query
authors = session.query(Author).options(selectinload(Author.books)).all()

Accessing a lazy relationship outside a session raises DetachedInstanceError

Question 9 / 15

How does connection pooling work in SQLAlchemy?

SQLAlchemy's create_engine() automatically manages a connection pool -- a cache of database connections that are reused rather than created and destroyed for every operation.

The default pool is QueuePool, which maintains a fixed-size pool of connections.

Key pool parameters:

pool_size=5 - Number of persistent connections to keep in the pool
max_overflow=10 - Extra connections allowed beyond pool_size (temporary)
pool_timeout=30 - Seconds to wait for a connection before raising an error
pool_recycle=3600 - Seconds after which a connection is recycled (important for MySQL which closes idle connections after 8 hours)
pool_pre_ping=True - Test connections before using them (handles stale connections)

Pool types:

QueuePool - Default, fixed-size with overflow
NullPool - No pooling, new connection every time (good for multiprocessing)
StaticPool - Single connection reused (good for testing with in-memory SQLite)
SingletonThreadPool - One connection per thread

Pseudo Code

1. create_engine() creates a pool automatically
2. Configure pool_size, max_overflow, pool_recycle
3. Use pool_pre_ping=True for production reliability
4. Use NullPool for multiprocessing scenarios
5. Use StaticPool for in-memory SQLite testing

Code

from sqlalchemy import create_engine, event
from sqlalchemy.pool import QueuePool, NullPool, StaticPool

# Production configuration
engine = create_engine(
    'postgresql://user:pass@localhost/mydb',
    pool_size=10,           # 10 persistent connections
    max_overflow=20,        # Up to 30 total (10 + 20 overflow)
    pool_timeout=30,        # Wait 30s for a connection
    pool_recycle=1800,      # Recycle connections after 30 minutes
    pool_pre_ping=True,     # Test connections before use
)

# Check pool status
print(engine.pool.status())  # Pool size: 10 Connections in pool: 0

# NullPool for multiprocessing (no shared connections)
from sqlalchemy.pool import NullPool

engine_mp = create_engine(
    'postgresql://user:pass@localhost/mydb',
    poolclass=NullPool,     # No pooling
)

# StaticPool for testing with in-memory SQLite
engine_test = create_engine(
    'sqlite://',
    poolclass=StaticPool,                   # Same connection always
    connect_args={'check_same_thread': False},  # Allow cross-thread
)

# Listening to pool events
@event.listens_for(engine, 'checkout')
def on_checkout(dbapi_conn, connection_rec, connection_proxy):
    print('Connection checked out from pool')

@event.listens_for(engine, 'checkin')
def on_checkin(dbapi_conn, connection_rec):
    print('Connection returned to pool')

# Explicit connection management
with engine.connect() as conn:
    result = conn.execute(text('SELECT 1'))
    # Connection returned to pool when context manager exits

# Dispose all pooled connections (e.g., after fork)
engine.dispose()

Key Points

•QueuePool is the default with pool_size=5 and max_overflow=10
•pool_pre_ping=True prevents stale connection errors in production
•pool_recycle is critical for MySQL which drops idle connections after 8 hours
•Use NullPool with multiprocessing to avoid shared connection issues
•engine.dispose() drops all pooled connections (useful after process fork)

Gotchas

•Forking a process with active pooled connections causes corruption; call engine.dispose() after fork
•SQLite in-memory databases need StaticPool or they create separate databases per connection
•pool_recycle should be less than the database server's connection timeout (e.g., MySQL wait_timeout)

SQLAlchemy automatically pools connections via QueuePool (default). Configure with pool_size, max_overflow, pool_recycle, and pool_pre_ping=True in create_engine(). Use NullPool for multiprocessing and StaticPool for in-memory SQLite tests.

•Default QueuePool reuses connections; pool_pre_ping=True prevents stale connections
•NullPool for multiprocessing, StaticPool for in-memory SQLite

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    pool_size=10,
    max_overflow=20,
    pool_recycle=1800,
    pool_pre_ping=True,
)

Always call engine.dispose() after forking a process to avoid connection corruption

Question 10 / 15

How do you execute raw SQL in SQLAlchemy?

Sometimes you need to execute raw SQL for complex queries, database-specific features, or performance. SQLAlchemy provides several ways:

text() - Wraps a raw SQL string as an executable object. Supports named parameters with :param syntax.
engine.connect() - Execute raw SQL through a connection.
session.execute() - Execute raw SQL within an ORM session (participates in the session's transaction).
from_statement() - Map raw SQL results to ORM models.

Always use text() with bound parameters (:param_name) instead of f-strings or string concatenation to prevent SQL injection.

Pseudo Code

1. Import text from sqlalchemy
2. Write SQL with :named_param placeholders
3. Execute via session.execute(text(sql), params)
4. Fetch results with .fetchall(), .fetchone(), .mappings()
5. For ORM mapping, use from_statement(text(sql))

Code

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

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

# Method 1: Via engine connection
with engine.connect() as conn:
    # Simple query
    result = conn.execute(text('SELECT * FROM users WHERE age > :age'), {'age': 18})
    rows = result.fetchall()
    for row in rows:
        print(row.name, row.age)  # Access by column name

    # As dictionaries
    result = conn.execute(text('SELECT * FROM users'))
    for row in result.mappings():
        print(row['name'], row['age'])

    # INSERT/UPDATE/DELETE
    conn.execute(
        text('INSERT INTO users (name, age) VALUES (:name, :age)'),
        {'name': 'Alice', 'age': 30}
    )
    conn.commit()  # Must commit with engine.connect()

# Method 2: Via ORM session
with Session(engine) as session:
    result = session.execute(
        text('SELECT * FROM users WHERE name = :name'),
        {'name': 'Alice'}
    )
    user_row = result.fetchone()
    print(user_row.name)

    # Bulk insert with executemany
    session.execute(
        text('INSERT INTO users (name, age) VALUES (:name, :age)'),
        [{'name': 'Bob', 'age': 25}, {'name': 'Charlie', 'age': 35}]
    )
    session.commit()

# Method 3: Map results to ORM models
with Session(engine) as session:
    stmt = text('SELECT id, name, email FROM users WHERE age > :age')
    users = session.query(User).from_statement(stmt).params(age=18).all()
    for user in users:
        print(user.name)  # Returns User objects, not raw rows

# Method 4: text() with type coercion
from sqlalchemy import Integer
stmt = text('SELECT id, name FROM users').columns(id=Integer)
result = session.execute(stmt)

Key Points

•Always use text() with :named_params for safety (never f-strings)
•session.execute() participates in the session transaction; engine.connect() does not
•.fetchall() returns a list of Row objects; .mappings() returns dict-like objects
•from_statement() maps raw SQL results to ORM model objects
•Pass a list of dicts to execute() for bulk operations

Gotchas

•engine.connect() requires explicit conn.commit(); session.execute() uses session.commit()
•Column names in raw SQL must match model attributes for from_statement() to work
•Never use f-strings or .format() in SQL strings -- always use :param_name for injection prevention

Use text() to wrap raw SQL and pass parameters with :name syntax. Execute via session.execute(text(sql), params) or engine.connect(). Use from_statement() to map results back to ORM models.

•Use text() with :named_params, never f-strings or concatenation
•session.execute() for ORM context; engine.connect() for raw context

from sqlalchemy import text

with Session(engine) as session:
    result = session.execute(
        text('SELECT * FROM users WHERE age > :age'),
        {'age': 18}
    )
    for row in result.fetchall():
        print(row.name)

Never interpolate user input directly into SQL strings

Question 11 / 15

How do you use Alembic for database migrations with SQLAlchemy?

Alembic is the official database migration tool for SQLAlchemy. It tracks schema changes (adding/removing tables, columns, indexes) in versioned migration scripts, allowing you to evolve your database schema incrementally and rollback if needed.

Key concepts:

Migration script: A Python file with upgrade() and downgrade() functions.
Revision: A unique identifier for each migration.
alembic_version table: Tracks the current migration version in the database.
Autogenerate: Alembic can compare your models to the database and generate migration scripts automatically.

Workflow:

alembic init migrations - Initialize Alembic in your project.
Edit alembic.ini and env.py to configure your database URL and model metadata.
alembic revision --autogenerate -m "message" - Generate a migration.
Review the generated script.
alembic upgrade head - Apply all pending migrations.
alembic downgrade -1 - Roll back one migration.

Pseudo Code

1. alembic init migrations (creates directory structure)
2. Configure sqlalchemy.url in alembic.ini
3. Import target_metadata in env.py from your models
4. alembic revision --autogenerate -m 'add users table'
5. Review the generated migration file
6. alembic upgrade head (apply)
7. alembic downgrade -1 (rollback if needed)

Code

# Terminal commands:
# pip install alembic
# alembic init migrations

# --- env.py (key parts) ---
from app.models import Base  # Your declarative base
target_metadata = Base.metadata

# --- alembic.ini (key line) ---
# sqlalchemy.url = sqlite:///app.db

# --- Generate a migration ---
# alembic revision --autogenerate -m "create users table"

# --- Generated migration file (example) ---
"""create users table

Revision ID: a1b2c3d4e5f6
Revises:
Create Date: 2024-01-15 10:30:00.000000
"""
from alembic import op
import sqlalchemy as sa

revision = 'a1b2c3d4e5f6'
down_revision = None
branch_labels = None
depends_on = None

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(100), nullable=False),
        sa.Column('email', sa.String(120), unique=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index('ix_users_email', 'users', ['email'])

def downgrade():
    op.drop_index('ix_users_email', 'users')
    op.drop_table('users')

# --- Common Alembic operations ---
# Add a column
def upgrade():
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

def downgrade():
    op.drop_column('users', 'age')

# Rename a column
def upgrade():
    op.alter_column('users', 'name', new_column_name='full_name')

# Terminal: Apply and manage migrations
# alembic upgrade head          # Apply all
# alembic downgrade -1          # Rollback one
# alembic history               # Show migration history
# alembic current               # Show current version
# alembic stamp head            # Mark DB as up-to-date without running migrations

Key Points

•Alembic tracks schema changes as versioned migration scripts with upgrade/downgrade
•--autogenerate compares models to the database and generates the diff
•Always review autogenerated migrations before applying them
•alembic upgrade head applies all pending migrations; downgrade -1 rolls back one
•target_metadata in env.py must point to your Base.metadata for autogenerate to work

Gotchas

•Autogenerate does not detect table renames, column type changes, or constraint name changes
•Running downgrade in production is risky if data was already written to new columns
•Always run migrations in a transaction; some databases (MySQL) do not support transactional DDL
•Forgetting to import all model modules before autogenerate results in empty migrations

Alembic tracks SQLAlchemy schema changes as migration scripts. Initialize with alembic init, configure env.py with your Base.metadata, generate migrations with alembic revision --autogenerate -m 'message', and apply with alembic upgrade head.

•autogenerate compares models to DB and creates migration scripts
•Always review generated migrations before applying

# Setup
# alembic init migrations
# Configure env.py: target_metadata = Base.metadata

# Generate migration
# alembic revision --autogenerate -m "add users table"

# Apply
# alembic upgrade head

# Rollback
# alembic downgrade -1

Autogenerate cannot detect renames or some type changes; review the output carefully

Question 12 / 15

How do you manage transactions in SQLAlchemy?

A transaction groups multiple database operations into an atomic unit: either all succeed or all are rolled back.

SQLAlchemy provides transactions at two levels:

Session-level (ORM): The Session automatically manages a transaction. Call session.commit() to persist, session.rollback() to undo. Using Session as a context manager auto-rolls back on exceptions.
Connection-level (Core): engine.connect() starts a transaction implicitly. Use conn.commit() or conn.rollback(). engine.begin() auto-commits on success and auto-rolls back on exception.

Nested transactions (savepoints): Use session.begin_nested() to create a savepoint within a transaction. You can roll back to the savepoint without rolling back the entire transaction.

AUTOCOMMIT mode: In SQLAlchemy 2.0, connections start in "autobegin" mode. Each operation begins a transaction automatically, but you must explicitly commit.

Pseudo Code

1. Use Session as context manager for auto-rollback on exception
2. session.commit() to save, session.rollback() to undo
3. engine.begin() for auto-commit/rollback at connection level
4. session.begin_nested() for savepoints
5. Wrap multi-step operations in try/except with rollback

Code

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

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

# Session-level transaction (ORM)
with Session(engine) as session:
    try:
        user1 = User(name='Alice')
        user2 = User(name='Bob')
        session.add_all([user1, user2])

        # More operations...
        session.execute(text('UPDATE accounts SET balance = balance - 100 WHERE id = 1'))
        session.execute(text('UPDATE accounts SET balance = balance + 100 WHERE id = 2'))

        session.commit()  # All or nothing
    except Exception:
        session.rollback()  # Undo everything
        raise

# Auto-rollback with context manager
with Session(engine) as session:
    with session.begin():  # Auto commits on exit, rollback on exception
        session.add(User(name='Alice'))
        session.add(User(name='Bob'))
        # No need to call session.commit()

# Connection-level transaction (Core)
with engine.begin() as conn:  # Auto-commit on success, rollback on exception
    conn.execute(text('INSERT INTO users (name) VALUES (:name)'), {'name': 'Alice'})
    conn.execute(text('INSERT INTO users (name) VALUES (:name)'), {'name': 'Bob'})

# Nested transaction (savepoint)
with Session(engine) as session:
    session.add(User(name='Alice'))  # This will be committed

    try:
        with session.begin_nested():  # SAVEPOINT
            session.add(User(name='Bob'))  # This might fail
            session.add(User(name='DuplicateEmail'))  # Causes IntegrityError
    except Exception:
        pass  # Savepoint is rolled back, but Alice is still pending

    session.commit()  # Only Alice is committed

Key Points

•Session context manager auto-rolls back on unhandled exceptions
•session.begin() auto-commits on success; no explicit commit() needed
•engine.begin() provides auto-commit/rollback at the connection level
•begin_nested() creates a savepoint for partial rollback within a transaction
•SQLAlchemy 2.0 uses autobegin: transactions start automatically but require explicit commit

Gotchas

•After a rollback, all objects loaded in that session become expired and may raise DetachedInstanceError
•Calling session.commit() inside a session.begin() block causes an error; let the context manager handle it
•SQLite does not support true nested transactions (savepoints work differently)

Use Session as a context manager for auto-rollback on exceptions. session.commit() to persist, session.rollback() to undo. session.begin() auto-commits on success. Use session.begin_nested() for savepoints (partial rollback).

•session.begin() auto-commits; session.begin_nested() creates a savepoint
•Context managers handle rollback on exceptions automatically

# Auto-commit/rollback
with Session(engine) as session:
    with session.begin():
        session.add(User(name='Alice'))
        session.add(User(name='Bob'))
        # Commits automatically; rolls back on exception

# Savepoint
with Session(engine) as session:
    session.add(User(name='Alice'))
    try:
        with session.begin_nested():
            session.add(User(name='Bad'))  # fails
    except Exception:
        pass
    session.commit()  # Only Alice is saved

After rollback, session objects are expired and may not be usable

Question 13 / 15

How do you use SQLAlchemy Sessions and the connection context manager pattern?

The Session is the ORM's interface to the database. It manages the identity map (cache of loaded objects), tracks changes, and handles transactions.

Session lifecycle patterns:

Context manager (recommended): with Session(engine) as session: automatically closes the session on exit.
sessionmaker: A factory that creates configured Session instances.
scoped_session: Thread-local Session registry (common in web apps). Each thread gets its own Session.

Key Session behaviors:

Identity map: If you query the same row twice, you get the same Python object.
Dirty tracking: SQLAlchemy tracks which objects have been modified.
Flush vs Commit: flush() writes pending changes to the DB (within the transaction). commit() flushes + commits the transaction.
Expunge: session.expunge(obj) removes an object from the session so it can live independently.

Pseudo Code

1. Create engine
2. Create sessionmaker bound to engine
3. Use SessionLocal() as context manager
4. For web apps, use scoped_session for thread-safety
5. Always close sessions after use

Code

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker, scoped_session

engine = create_engine('sqlite:///app.db')

# Pattern 1: Direct Session context manager
with Session(engine) as session:
    user = session.query(User).first()
    user.name = 'Updated'
    session.commit()
# Session is closed automatically

# Pattern 2: sessionmaker factory
SessionLocal = sessionmaker(bind=engine)

with SessionLocal() as session:
    session.add(User(name='Alice'))
    session.commit()

# Pattern 3: sessionmaker without context manager
session = SessionLocal()
try:
    session.add(User(name='Bob'))
    session.commit()
except Exception:
    session.rollback()
    raise
finally:
    session.close()

# Pattern 4: scoped_session (thread-local, common in Flask/web apps)
db_session = scoped_session(sessionmaker(bind=engine))

# Each thread gets its own session
user = db_session.query(User).first()  # Uses thread-local session
db_session.commit()
db_session.remove()  # Clean up after the request

# Flask-style pattern with dependency injection
def get_db():
    """Yield a database session, ensuring cleanup."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Identity map demonstration
with Session(engine) as session:
    user1 = session.query(User).get(1)
    user2 = session.query(User).get(1)
    assert user1 is user2  # Same Python object!

# Flush vs Commit
with Session(engine) as session:
    user = User(name='Alice')
    session.add(user)
    session.flush()      # SQL INSERT runs, user.id is set, but NOT committed
    print(user.id)       # ID is available now
    session.commit()     # Transaction is committed

# Expunge: detach object from session
with Session(engine) as session:
    user = session.query(User).first()
    session.expunge(user)  # Detach from session
# user can be used outside the session now (attributes already loaded are accessible)

Key Points

•Session as context manager is the recommended pattern; auto-closes on exit
•sessionmaker creates a factory for configured sessions; scoped_session adds thread-local behavior
•flush() writes to DB within the transaction; commit() finalizes the transaction
•Identity map ensures the same DB row maps to the same Python object within a session
•scoped_session.remove() is required at the end of web requests to clean up

Gotchas

•Accessing relationships on an object after session.close() raises DetachedInstanceError
•session.flush() can raise IntegrityError even before commit() if constraints are violated
•scoped_session without remove() can leak sessions and exhaust the connection pool

Use Session(engine) as a context manager or sessionmaker(bind=engine) as a factory. flush() writes SQL without committing; commit() finalizes. Use scoped_session for thread-local sessions in web apps, always calling .remove() after each request.

•sessionmaker for factories, Session context manager for lifecycle
•flush() writes SQL, commit() finalizes the transaction

from sqlalchemy.orm import sessionmaker, Session

SessionLocal = sessionmaker(bind=engine)

# Recommended: context manager
with SessionLocal() as session:
    session.add(User(name='Alice'))
    session.commit()

# flush vs commit
with Session(engine) as session:
    user = User(name='Alice')
    session.add(user)
    session.flush()   # INSERT runs, user.id set, not yet committed
    session.commit()  # Finalized

Accessing lazy-loaded relationships after session.close() raises DetachedInstanceError

Question 14 / 15

How do you perform bulk insert, update, and delete operations in SQLAlchemy?

For large datasets, using session.add() in a loop is very slow because SQLAlchemy emits individual INSERT statements and tracks each object in the identity map.

Bulk operation strategies (fastest to slowest):

Core insert() - Bypasses the ORM entirely. Fastest.
session.bulk_insert_mappings() - Accepts dicts, skips identity map. Fast.
session.bulk_save_objects() - Accepts model instances, skips some tracking. Medium.
session.add_all() - Full ORM tracking. Slowest but safest.

For bulk updates and deletes, use .update() and .delete() on queries to generate a single UPDATE/DELETE statement instead of loading all objects into Python.

Pseudo Code

1. For inserts: prefer Core insert() or bulk_insert_mappings()
2. For updates: use session.query(Model).filter(...).update({...})
3. For deletes: use session.query(Model).filter(...).delete()
4. Set synchronize_session=False for best performance
5. Batch large operations into chunks

Code

from sqlalchemy import insert, update, delete
from sqlalchemy.orm import Session

# --- BULK INSERT ---

# Method 1: Core insert (fastest)
with Session(engine) as session:
    session.execute(
        insert(User),
        [
            {'name': 'Alice', 'age': 30},
            {'name': 'Bob', 'age': 25},
            {'name': 'Charlie', 'age': 35},
        ]
    )
    session.commit()

# Method 2: bulk_insert_mappings (fast, ORM-aware)
with Session(engine) as session:
    session.bulk_insert_mappings(User, [
        {'name': 'Alice', 'age': 30},
        {'name': 'Bob', 'age': 25},
    ])
    session.commit()

# Method 3: bulk_save_objects (uses model instances)
with Session(engine) as session:
    users = [User(name=f'User_{i}', age=20+i) for i in range(1000)]
    session.bulk_save_objects(users)
    session.commit()

# Method 4: add_all (slowest, full tracking)
with Session(engine) as session:
    session.add_all([User(name='Alice'), User(name='Bob')])
    session.commit()

# --- BULK UPDATE ---

# ORM bulk update (single UPDATE statement)
with Session(engine) as session:
    session.query(User).filter(User.age < 18).update(
        {'is_active': False},
        synchronize_session='fetch'  # or False for speed
    )
    session.commit()

# Core update
with Session(engine) as session:
    session.execute(
        update(User).where(User.age < 18).values(is_active=False)
    )
    session.commit()

# --- BULK DELETE ---

# ORM bulk delete
with Session(engine) as session:
    deleted_count = session.query(User).filter(User.is_active == False).delete(
        synchronize_session='fetch'
    )
    print(f'Deleted {deleted_count} users')
    session.commit()

# --- Chunked processing for very large datasets ---
def bulk_insert_chunked(session, data, chunk_size=1000):
    for i in range(0, len(data), chunk_size):
        chunk = data[i:i + chunk_size]
        session.bulk_insert_mappings(User, chunk)
        session.flush()  # Write chunk to DB
    session.commit()

Key Points

•Core insert() is the fastest; it bypasses the ORM identity map entirely
•bulk_insert_mappings() takes dicts and is much faster than add_all()
•.update() and .delete() on queries generate single SQL statements without loading objects
•synchronize_session='fetch' keeps the session in sync; False is faster but session may be stale
•Chunk large operations to avoid memory issues

Gotchas

•bulk_save_objects() and bulk_insert_mappings() skip relationship cascades and events
•synchronize_session=False means in-memory objects are not updated after bulk update/delete
•Bulk operations do not trigger ORM events (before_insert, after_update, etc.)

Use Core insert() or bulk_insert_mappings() for fast bulk inserts (bypasses ORM tracking). Use session.query(Model).filter(...).update({}) and .delete() for bulk updates/deletes without loading objects into Python.

•Core insert() and bulk_insert_mappings() bypass ORM overhead
•.update() and .delete() on queries execute single SQL statements

# Bulk insert (fast)
session.execute(insert(User), [{'name': 'Alice'}, {'name': 'Bob'}])
session.commit()

# Bulk update (single SQL UPDATE)
session.query(User).filter(User.age < 18).update({'is_active': False})
session.commit()

# Bulk delete
session.query(User).filter(User.is_active == False).delete()
session.commit()

Bulk operations skip ORM events and relationship cascades

Question 15 / 15

What are hybrid properties in SQLAlchemy and when do you use them?

A hybrid property is a descriptor that behaves differently at the Python (instance) level and the SQL (class/query) level:

At the instance level: Acts like a regular Python @property, computing a value from the instance's attributes.
At the class level (in queries): Generates SQL expressions so the computation happens in the database.

This is powerful because you can use the same attribute name for both Python logic and SQL filtering/ordering.

Use cases:

Computed fields (full name from first + last)
Status flags derived from dates or amounts
Case-insensitive lookups
Virtual columns that should be filterable

Pseudo Code

1. Import hybrid_property from sqlalchemy.ext.hybrid
2. Define a @hybrid_property method for instance-level computation
3. Define a @name.expression classmethod for SQL-level expression
4. Use in queries: session.query(Model).filter(Model.hybrid_attr == value)

Code

from sqlalchemy import Column, Integer, String, case, func
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method
from sqlalchemy.orm import declarative_base, Session

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    first_name = Column(String(50))
    last_name = Column(String(50))
    salary = Column(Integer)

    # Hybrid property: full_name
    @hybrid_property
    def full_name(self):
        """Python-level: concatenate strings."""
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    def full_name(cls):
        """SQL-level: use || or concat."""
        return cls.first_name + ' ' + cls.last_name

    # Hybrid property with setter
    @hybrid_property
    def email_lower(self):
        return self._email.lower() if self._email else None

    # Hybrid property for status
    @hybrid_property
    def salary_tier(self):
        if self.salary >= 100000:
            return 'senior'
        elif self.salary >= 50000:
            return 'mid'
        return 'junior'

    @salary_tier.expression
    def salary_tier(cls):
        return case(
            (cls.salary >= 100000, 'senior'),
            (cls.salary >= 50000, 'mid'),
            else_='junior'
        )

# Usage in queries
with Session(engine) as session:
    # Filter using hybrid property (generates SQL)
    user = session.query(User).filter(User.full_name == 'John Doe').first()

    # Order by hybrid property
    users = session.query(User).order_by(User.full_name).all()

    # Filter by salary tier
    seniors = session.query(User).filter(User.salary_tier == 'senior').all()

    # Use on an instance (runs Python)
    user = session.query(User).first()
    print(user.full_name)       # "Alice Smith" (Python)
    print(user.salary_tier)     # "senior" (Python)

Key Points

•hybrid_property works at both Python (instance) and SQL (class) levels
•The @expression classmethod defines the SQL version used in filter/order_by
•Without an @expression, the property works on instances but not in queries
•Use case() for SQL CASE WHEN expressions in the SQL-level definition
•Hybrid properties keep business logic DRY: same name for Python and SQL

Gotchas

•Without the @expression classmethod, filtering/ordering by the hybrid property raises an error
•SQL expressions may behave differently than Python (e.g., NULL handling, string concatenation)
•Complex hybrid expressions can generate slow SQL; check the query plan with echo=True

A hybrid_property works as a Python @property on instances and as a SQL expression in queries. Define the Python version with @hybrid_property and the SQL version with @name.expression. This lets you filter/order by computed attributes in the database.

•@hybrid_property for Python; @name.expression for SQL
•Allows filtering and ordering by computed attributes

from sqlalchemy.ext.hybrid import hybrid_property

class User(Base):
    __tablename__ = 'users'
    first_name = Column(String(50))
    last_name = Column(String(50))

    @hybrid_property
    def full_name(self):
        return f"{self.first_name} {self.last_name}"

    @full_name.expression
    def full_name(cls):
        return cls.first_name + ' ' + cls.last_name

# Works in queries: User.full_name == 'John Doe' generates SQL

Without @expression, the property only works on instances, not in queries

Built by onllm.dev

Home GitHub Support awesomePrep