Creating partial and expression indexes from SQLAlchemy models

Declare partial indexes with Index("ix_jobs_pending", "run_after", postgresql_where=text("status = 'pending'")) and expression indexes with Index("ix_customers_email_lower", func.lower(Customer.email)) in the model — so autogenerate keeps them, and the migration only has to decide whether to build them concurrently. This guide belongs to managing enums, constraints and indexes in migrations.

Quick Answer

Two index shapes cover most of the cases a plain column index handles badly: a predicate that restricts which rows are stored, and an expression that stores a computed value.

Before — full indexes that queries cannot use or do not need:

import datetime as dt

from sqlalchemy import Index, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Job(Base):
    __tablename__ = "jobs"
    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str]
    run_after: Mapped[dt.datetime]
    __table_args__ = (Index("ix_jobs_status", "status"),)   # 99% of rows are 'complete'


# And a case-insensitive lookup that cannot use an index at all:
select(Customer).where(func.lower(Customer.email) == email.lower())   # Seq Scan

After — a partial index and an expression index:

import datetime as dt

from sqlalchemy import Index, func, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(primary_key=True)
    status: Mapped[str]
    run_after: Mapped[dt.datetime]

    __table_args__ = (
        # Only the rows a worker ever looks for.
        Index(
            "ix_jobs_pending_run_after",
            "run_after",
            postgresql_where=text("status = 'pending'"),
        ),
    )


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    deleted_at: Mapped[dt.datetime | None] = mapped_column(default=None)


# Declared after the class, because the expression needs the mapped attribute.
Index("ix_customers_email_lower", func.lower(Customer.email))

# Unique among live rows only.
Index(
    "uq_customers_email_live",
    Customer.email,
    unique=True,
    postgresql_where=Customer.deleted_at.is_(None),
)

The partial index holds a few thousand rows instead of millions, and the expression index is what turns the case-insensitive lookup from a sequential scan into an index scan.

Execution Context & Async Workflow Integration

A partial index stores only rows satisfying its predicate. That makes it smaller — often by orders of magnitude for a status column with a skewed distribution — faster to scan, and cheaper to maintain, because an update to a row outside the predicate does not touch the index at all.

The condition for using it is that the planner must be able to prove the query's predicate implies the index's. WHERE status = 'pending' AND run_after <= now() can use an index restricted to status = 'pending'. WHERE run_after <= now() alone cannot, even if every matching row happens to be pending. That is why a partial index and the query it serves have to be designed together: the predicate is part of the contract.

An expression index stores the result of a function or expression, and the matching rule is stricter still — the query's expression must be textually equivalent to the indexed one after parsing. lower(email) matches lower(email); it does not match upper(email), lower(email || ''), or a query that applies lower() to the parameter instead of the column. The expression must also be IMMUTABLE: lower() qualifies, now() does not, and a function of your own needs to be declared immutable to be indexable.

In SQLAlchemy, both are declared in the model so that autogenerate compares them. postgresql_where takes a SQL expression — either a text() fragment or a SQLAlchemy expression such as Customer.deleted_at.is_(None) — and expression indexes take the expression directly, which means they have to be declared after the class, because mapped_column() inside the class body is a placeholder rather than a usable attribute.

Two SQLAlchemy-specific notes. postgresql_include=[...] adds non-key columns to the index so a query can be answered from the index alone, which is PostgreSQL's covering-index support. And postgresql_using="gin" or "gist" selects the index type, which matters for the JSONB and full-text cases in querying Postgres JSONB, arrays and full-text search.

postgresql_concurrently is deliberately not declared in the model. It describes how an index is built, not what the index is, and autogenerate does not compare it — so it belongs only in the migration, wrapped in an autocommit block as creating indexes concurrently in Alembic migrations describes.

Resolving Warnings, Errors & Common Mistakes

The repeated-recreation case is the one most likely to be met in practice. PostgreSQL normalises index predicates — status = 'pending' becomes (status = 'pending'::text) — and Alembic compares the text, so a predicate written differently in the model produces a difference on every autogenerate run. Two workable responses: write the predicate the way PostgreSQL reports it, which can be read from pg_indexes, or keep the index in the model for documentation and exclude it from comparison with include_object.

Reading back what the database actually has is a one-line query and settles the question:

from sqlalchemy import text

INDEX_DEFINITIONS = text("""
    SELECT indexname, indexdef
    FROM pg_indexes
    WHERE schemaname = 'public' AND tablename = :table
    ORDER BY indexname
""")

The other frequent surprise is a partial index that could be used but is not, because the planner estimates a sequential scan to be cheaper. That is often correct for a small table, and worth verifying rather than arguing with — the EXPLAIN workflow in reading EXPLAIN output for a SQLAlchemy query shows both the chosen plan and the estimated costs, so the comparison is visible.

For the unique-constraint case, note that a partial unique index is not a constraint: it cannot be the target of a foreign key, and it cannot be named in ON CONFLICT (...) by constraint name — though ON CONFLICT with the same column list and predicate does use it. Where a real constraint is needed, the full constraint has to stay, and the partial rule has to be expressed some other way.

Advanced: Partial Unique Indexes and Covering Indexes

Two shapes solve problems a plain index cannot express at all.

A partial unique index enforces uniqueness over a subset of rows. The canonical case is "one active row per owner", which no plain unique constraint can state:

from sqlalchemy import Boolean, ForeignKey, Index, text
from sqlalchemy.orm import Mapped, mapped_column


class Address(Base):
    __tablename__ = "addresses"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))
    is_default: Mapped[bool] = mapped_column(Boolean, default=False)

    __table_args__ = (
        # At most one default address per customer; any number of non-default ones.
        Index(
            "uq_addresses_one_default_per_customer",
            "customer_id",
            unique=True,
            postgresql_where=text("is_default"),
        ),
    )

That single line replaces application logic that would otherwise have to check-then-write under a lock, and it is enforced for every writer. Setting a new default becomes two statements in one transaction — clear the old, set the new — and the index guarantees the invariant even under concurrency, in the same way described in handling IntegrityError on concurrent inserts.

A covering index adds non-key columns so a query can be satisfied without reading the table at all:

from sqlalchemy import Index

Index(
    "ix_orders_customer_placed_covering",
    Order.customer_id,
    Order.placed_at.desc(),
    postgresql_include=["status", "total_cents"],
)

A query selecting status and total_cents for one customer, ordered by date, can then be an index-only scan: PostgreSQL reads the index and never touches the heap. The gain is real for hot list endpoints and comes at the cost of a larger index and more work per write, so it is worth adding with a measurement rather than by default.

Combining the shapes is often the most valuable version. A partial covering index for a job queue holds only pending jobs and carries everything the claim query reads:

Index(
    "ix_jobs_pending_claim",
    Job.run_after,
    postgresql_where=text("status = 'pending'"),
    postgresql_include=["id", "payload"],
)

For a table where pending rows are a tiny fraction of the total, that index stays small enough to remain in cache permanently — which is what makes a queue on a large table behave like a queue on a small one, and the reason it appears in building a job queue with SELECT FOR UPDATE SKIP LOCKED.

Verifying the Index Is Actually Used

An index that the planner ignores costs write throughput and buys nothing, and the only way to know which you have is to look. Three checks cover it.

Read the plan for the exact query. Compile the statement SQLAlchemy will send, with its parameters, and run EXPLAIN on it. What you want to see is the index by name:

from sqlalchemy import text


async def explain(session, stmt) -> str:
    compiled = stmt.compile(session.bind, compile_kwargs={"literal_binds": True})
    rows = await session.execute(text(f"EXPLAIN (ANALYZE, BUFFERS) {compiled}"))
    return "\n".join(row[0] for row in rows)

For a partial index, also check the negative case: the same query without the predicate should fall back to a scan. If it does not, the index is wider than you thought.

Check usage counts after a week. pg_stat_user_indexes.idx_scan counts how often each index has been used. An index at zero after real traffic is serving nothing:

from sqlalchemy import text

UNUSED = text("""
    SELECT relname AS table_name, indexrelname AS index_name,
           idx_scan, pg_size_pretty(pg_relation_size(indexrelid)) AS size
    FROM pg_stat_user_indexes
    WHERE idx_scan = 0
    ORDER BY pg_relation_size(indexrelid) DESC
""")

Check replicas too before dropping anything: statistics are per server, and an index idle on the primary may be busy serving read traffic elsewhere.

Assert the definition in a test. For indexes that matter — a partial unique index enforcing an invariant, in particular — a test that reads pg_indexes and compares the definition catches a migration that dropped or altered it:

import pytest
from sqlalchemy import text


@pytest.mark.asyncio
async def test_one_default_address_per_customer(session, customer_factory, address_factory):
    from sqlalchemy.exc import IntegrityError

    customer_id = await customer_factory()
    await address_factory(customer_id=customer_id, is_default=True)
    await session.commit()

    await address_factory(customer_id=customer_id, is_default=True)
    with pytest.raises(IntegrityError):
        await session.commit()

That test asserts the behaviour rather than the definition, which is better: it keeps passing if the index is renamed and fails if the invariant stops being enforced — which is the thing that actually matters. Running it against a schema built by the migrations, as in running tests against a Postgres testcontainer, is what makes it meaningful.

Frequently Asked Questions

Related

• Managing Enums, Constraints and Indexes in Migrations — The parent guide: schema objects autogenerate handles poorly.
• Creating indexes concurrently in Alembic migrations — Building these indexes without blocking writes.
• Naming constraints with a metadata naming convention — Predictable names so autogenerate can compare them.
• Filtering soft-deleted rows with with_loader_criteria — The queries a partial index on deleted_at serves.