Implementing full-text search with tsvector in SQLAlchemy

Store a weighted TSVECTOR in a generated column, index it with GIN, and filter with Product.search_vector.bool_op("@@")(func.websearch_to_tsquery("english", q)) — then rank with ts_rank_cd and limit before you highlight. This guide belongs to querying PostgreSQL JSONB, arrays and full-text search.

Quick Answer

The slow way computes the document vector on every search, and the fragile way passes raw user input to to_tsquery. A stored, indexed vector and a forgiving parser fix both.

Before — vector computed per query, parser that raises on user input:

from sqlalchemy import func, select

from shop.models import Product


def search_products(q: str):
    document = func.to_tsvector("english", Product.name + " " + Product.description)
    return select(Product).where(document.bool_op("@@")(func.to_tsquery("english", q)))

# search_products("usb c cable") ->
# ProgrammingError: syntax error in tsquery: "usb c cable"

After — a generated, weighted vector, a GIN index and websearch syntax:

from sqlalchemy import Computed, Index, func, select
from sqlalchemy.dialects.postgresql import TSVECTOR
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    description: Mapped[str] = mapped_column(default="")
    search_vector: Mapped[str] = mapped_column(
        TSVECTOR,
        Computed(
            "setweight(to_tsvector('english', coalesce(name, '')), 'A') || "
            "setweight(to_tsvector('english', coalesce(description, '')), 'B')",
            persisted=True,
        ),
    )

    __table_args__ = (
        Index("ix_products_search_vector", "search_vector", postgresql_using="gin"),
    )


def search_products(q: str, limit: int = 20):
    query = func.websearch_to_tsquery("english", q)
    rank = func.ts_rank_cd(Product.search_vector, query).label("rank")
    return (
        select(Product, rank)
        .where(Product.search_vector.bool_op("@@")(query))
        .order_by(rank.desc(), Product.id)
        .limit(limit)
    )

websearch_to_tsquery accepts what people type into search boxes — quoted phrases, or, a leading minus to exclude — and never raises on malformed input, so the endpoint cannot be broken by a stray colon.

Execution Context & Async Workflow Integration

Full-text search in PostgreSQL has two halves, and SQLAlchemy mostly stays out of the way of both. A tsvector is a normalised, sorted list of lexemes with positions and weights: to_tsvector('english', 'Wireless Mice') becomes 'mice':2 'wireless':1, with stemming applied and stop words dropped. A tsquery is a boolean expression over lexemes. The @@ operator asks whether a vector satisfies a query, and a GIN index over vectors answers that for a whole table by looking up each lexeme.

The generated column is where the work moves from read time to write time. Computed(..., persisted=True) emits GENERATED ALWAYS AS (...) STORED, so PostgreSQL recomputes the vector whenever name or description changes, and the ORM treats the column as server-generated — it never writes to it, and after an insert or update it expires the attribute so the next read fetches the fresh value. Two rules make the expression acceptable to PostgreSQL: it must be immutable, which is why the two-argument to_tsvector('english', ...) form is required (the one-argument form depends on a session setting), and it may only reference columns of the same row.

On the read side, SQLAlchemy's .match() renders @@ plainto_tsquery(...), which ANDs every word and ignores operators. That is adequate for simple cases; for a search box, bool_op("@@") with func.websearch_to_tsquery gives users the syntax they expect. Either way, the search text travels as a bound parameter through asyncpg, so there is no injection surface in the SQL itself.

Under async, nothing about the query needs special handling — it is an ordinary select() awaited through session.execute(). The part that does need thought is result size. ts_rank_cd must evaluate every matching row to sort them, so a query that matches half the catalogue ranks half the catalogue. Keep LIMIT in the statement, add a cheap pre-filter where the domain has one (category, availability), and apply keyset pagination on (rank, id) rather than OFFSET for deep pages.

Resolving Warnings, Errors & Common Mistakes

The NULL concatenation row catches many first implementations, because it fails silently — the rows are simply never found. coalesce() each source column separately, as the generated expression above does, and use setweight per column rather than concatenating text, which also keeps the weights meaningful.

Changing the generated expression later is a migration concern. Alembic's autogenerate does not detect a change to a Computed expression; it warns and skips. Write the migration by hand — drop and re-add the column, or for large tables follow the staged approach below — and review it like any autogenerated migration script.

Advanced: Ranking, Snippets and Prefix Matching

Ranking is where search starts to feel good or bad, and it has three levers: weights, normalisation and what you do with ties.

Weights come from setweight in the generated column. ts_rank_cd multiplies each matched lexeme's contribution by its weight, with a default array of {0.1, 0.2, 0.4, 1.0} for D, C, B and A. A product whose name contains the query therefore outranks one that mentions it only in the description. Pass a custom array as the first argument to change the balance, and a normalisation flag as the last to stop long descriptions accumulating score merely by being long:

from sqlalchemy import func, literal_column, select

from shop.db import Session
from shop.models import Product

WEIGHTS = literal_column("'{0.05, 0.1, 0.5, 1.0}'::float4[]")


async def search_page(q: str, limit: int = 20) -> list[dict]:
    query = func.websearch_to_tsquery("english", q)
    # 32 = rank / (rank + 1): scales into 0..1 and damps very long documents.
    rank = func.ts_rank_cd(WEIGHTS, Product.search_vector, query, 32).label("rank")

    page = (
        select(Product.id, Product.name, Product.description, rank)
        .where(Product.search_vector.bool_op("@@")(query))
        .order_by(rank.desc(), Product.id)
        .limit(limit)
        .subquery()
    )
    # Headline only the rows on this page; ts_headline re-parses the source text.
    snippet = func.ts_headline(
        "english", page.c.description, query,
        "MaxWords=24, MinWords=12, StartSel=<mark>, StopSel=</mark>",
    ).label("snippet")

    async with Session() as session:
        rows = await session.execute(
            select(page.c.id, page.c.name, page.c.rank, snippet).order_by(
                page.c.rank.desc(), page.c.id
            )
        )
        return [dict(row._mapping) for row in rows]

ts_headline output contains the markers you asked for, and the surrounding text is your data, so escape the text and allow only the marker tags when rendering it in HTML.

Autocomplete is the one case where to_tsquery belongs, because you build the query yourself. Normalise the user's last word to letters and digits, and append :* for a prefix match:

import re

from sqlalchemy import func

_WORD = re.compile(r"[^\w]+", re.UNICODE)


def prefix_query(text: str):
    words = [w for w in _WORD.split(text.lower()) if w]
    if not words:
        return None
    terms = " & ".join(words[:-1] + [f"{words[-1]}:*"])
    return func.to_tsquery("english", terms)

For typo tolerance, full-text search is the wrong tool — stemming does not forgive misspellings. The pg_trgm extension with a GIN trigram index on name handles "wirless" matching "wireless", and a common design runs the tsvector search first and falls back to trigram similarity when it returns nothing.

Rolling Out Search on a Large Existing Table

The generated column in the quick answer is one statement to add, and on a small table that is the right way to do it. On a table with tens of millions of rows it is a problem: adding a stored generated column rewrites every row while holding an ACCESS EXCLUSIVE lock, which blocks reads and writes for the duration.

The staged alternative keeps the table online. Add an ordinary nullable tsvector column, which is a metadata-only change. Install a trigger that sets it on insert and update, so new writes are covered from that moment. Backfill existing rows in primary-key batches, committing each batch, so no single transaction holds locks for long or bloats the WAL. Then build the index with CREATE INDEX CONCURRENTLY, which in Alembic has to run outside the migration's transaction:

# alembic/versions/7c1e_products_search_index.py
from alembic import op


def upgrade() -> None:
    with op.get_context().autocommit_block():
        op.create_index(
            "ix_products_search_vector",
            "products",
            ["search_vector"],
            postgresql_using="gin",
            postgresql_concurrently=True,
        )


def downgrade() -> None:
    with op.get_context().autocommit_block():
        op.drop_index(
            "ix_products_search_vector", table_name="products", postgresql_concurrently=True
        )

A concurrent build that fails leaves an INVALID index behind, which PostgreSQL maintains on every write but never uses for reads. Check pg_index.indisvalid after the migration, and drop and rebuild if it is false.

In the ORM, a trigger-maintained column is mapped with server_default=FetchedValue() and server_onupdate=FetchedValue() instead of Computed, which tells SQLAlchemy the database sets it and the attribute must be re-read after a write. The same expand, backfill, switch sequence underlies most large-table changes, and adding a NOT NULL column without locking walks through its batching and verification in more detail.

Frequently Asked Questions

Related

• Querying Postgres JSONB, Arrays and Full-Text Search — The parent guide: JSONB, ARRAY and tsvector in one data model.
• Querying and indexing JSONB columns in SQLAlchemy — GIN indexing for documents, and choosing operators it can serve.
• Paginating large result sets with keyset pagination — Deep pages of ranked results without OFFSET.
• Creating indexes concurrently in Alembic migrations — Building the GIN index without blocking writes.