Module 7: Useful Extensions

Mini-project: refactoring a user system with extensions

Next →

You close out the module by applying 3 extensions (citext, gen_random_uuid, pg_trgm) to a real user system. You'll take a typical "before" schema and refactor it "after" using the right extensions. By the end you'll have a reusable pattern for production-ready user systems.

The "before" — typical schema without extensions

# app/models/user.py — initial version (improvable)
from datetime import datetime, timezone
from sqlalchemy import String, DateTime, Integer
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(200), unique=True, nullable=False)
    username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
    full_name: Mapped[str] = mapped_column(String(200))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

Problems:

  1. id is SERIAL — predictable, exposes count, not globally unique.
  2. email is VARCHAR — case-sensitive. Bug: a user signs up with John@Example.com, tries to log in with john@example.com, and isn't found.
  3. username is VARCHAR — same problem.
  4. No search — finding users by a similar name (typo) doesn't work.

The "after" — refactor with extensions

# app/models/user.py — refactor
import uuid
from datetime import datetime, timezone

from sqlalchemy import DateTime, String, text
from sqlalchemy.dialects.postgresql import UUID, CITEXT
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class User(Base):
    __tablename__ = "users"

    # 1. UUID — globally unique, doesn't expose count
    id: Mapped[uuid.UUID] = mapped_column(
        UUID(as_uuid=True),
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )

    # 2. CITEXT — automatic case-insensitive
    email: Mapped[str] = mapped_column(CITEXT, unique=True, nullable=False)
    username: Mapped[str] = mapped_column(CITEXT, unique=True, nullable=False)

    # 3. full_name plain VARCHAR — case matters for names
    full_name: Mapped[str] = mapped_column(String(200))

    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        default=lambda: datetime.now(timezone.utc),
    )

Plus an index for pg_trgm search (in the migration):

-- For fuzzy search of full_name
CREATE INDEX idx_users_fullname_trgm ON users USING gin (full_name gin_trgm_ops);

Complete migration

# alembic/versions/XXX_refactor_users.py
"""Refactor users with citext + UUID + pg_trgm

Changes:
- id: SERIAL → UUID with gen_random_uuid()
- email: VARCHAR → CITEXT
- username: VARCHAR → CITEXT
- Add GIN trigram index on full_name
"""

from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql


def upgrade() -> None:
    # Install extensions
    op.execute("CREATE EXTENSION IF NOT EXISTS citext")
    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
    # gen_random_uuid is built-in in PG 13+. For older PG: CREATE EXTENSION pgcrypto.

    # === For a new project: new table ===
    op.create_table(
        'users',
        sa.Column('id', postgresql.UUID(as_uuid=True),
                  server_default=sa.text("gen_random_uuid()"),
                  primary_key=True),
        sa.Column('email', postgresql.CITEXT(), unique=True, nullable=False),
        sa.Column('username', postgresql.CITEXT(), unique=True, nullable=False),
        sa.Column('full_name', sa.String(200), nullable=False),
        sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
    )

    # GIN trigram index
    op.execute("""
        CREATE INDEX idx_users_fullname_trgm
        ON users USING gin (full_name gin_trgm_ops)
    """)


def downgrade() -> None:
    op.drop_index('idx_users_fullname_trgm')
    op.drop_table('users')
    # Don't drop extensions — they may be used by other tables

Search endpoint with "did you mean"

# app/routers/users.py
from typing import Optional
import uuid
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.deps import get_db
from app.models import User


router = APIRouter()


class UserSearchResult(BaseModel):
    id: uuid.UUID
    email: str
    username: str
    full_name: str
    similarity: Optional[float] = None


class SearchResponse(BaseModel):
    results: list[UserSearchResult]
    did_you_mean: list[str] = []


@router.get("/users/search", response_model=SearchResponse)
async def search_users(
    q: str,
    db: AsyncSession = Depends(get_db),
):
    """Search users by full_name with smart suggestions."""
    if not q or len(q) < 2:
        raise HTTPException(400, "Query must be at least 2 chars")

    # Exact + prefix matches first
    exact_result = await db.execute(text("""
        SELECT id, email, username, full_name, 1.0 AS sim
        FROM users
        WHERE full_name ILIKE :pattern
        ORDER BY full_name
        LIMIT 10
    """), {"pattern": f"%{q}%"})

    exact_users = [
        UserSearchResult(
            id=r.id, email=r.email, username=r.username,
            full_name=r.full_name, similarity=r.sim
        )
        for r in exact_result.mappings()
    ]

    # If few exact matches, add fuzzy
    fuzzy_users = []
    if len(exact_users) < 3:
        fuzzy_result = await db.execute(text("""
            SELECT id, email, username, full_name,
                   similarity(full_name, :q) AS sim
            FROM users
            WHERE full_name % :q
              AND full_name NOT ILIKE :pattern
            ORDER BY sim DESC
            LIMIT 5
        """), {"q": q, "pattern": f"%{q}%"})

        fuzzy_users = [
            UserSearchResult(
                id=r.id, email=r.email, username=r.username,
                full_name=r.full_name, similarity=r.sim
            )
            for r in fuzzy_result.mappings()
        ]

    return SearchResponse(
        results=exact_users + fuzzy_users,
        did_you_mean=[u.full_name for u in fuzzy_users[:3]],
    )


@router.post("/users", response_model=UserSearchResult)
async def create_user(
    email: str,
    username: str,
    full_name: str,
    db: AsyncSession = Depends(get_db),
):
    """Create a user. citext handles case automatically."""
    user = User(email=email, username=username, full_name=full_name)
    db.add(user)
    try:
        await db.commit()
        await db.refresh(user)
    except Exception as e:
        await db.rollback()
        raise HTTPException(409, f"User already exists: {e}")

    return UserSearchResult(
        id=user.id, email=user.email, username=user.username,
        full_name=user.full_name
    )


@router.get("/users/by-email/{email}", response_model=UserSearchResult)
async def get_user_by_email(
    email: str,
    db: AsyncSession = Depends(get_db),
):
    """Search by email. citext: automatic case-insensitive."""
    user = await db.scalar(select(User).where(User.email == email))
    if not user:
        raise HTTPException(404, "User not found")

    return UserSearchResult(
        id=user.id, email=user.email, username=user.username,
        full_name=user.full_name
    )

Tests for the refactor

# tests/test_users_refactored.py
import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_email_case_insensitive(client: AsyncClient):
    """citext: emails case-insensitive."""
    # Create with one case
    r1 = await client.post("/users", json={
        "email": "John@Example.com",
        "username": "John",
        "full_name": "John Doe"
    })
    assert r1.status_code == 200

    # Search with a different case
    r2 = await client.get("/users/by-email/JOHN@example.com")
    assert r2.status_code == 200
    assert r2.json()["email"] == "John@Example.com"  # original case preserved


@pytest.mark.asyncio
async def test_email_case_insensitive_unique(client: AsyncClient):
    """citext: inserting a duplicate with a different case fails."""
    r1 = await client.post("/users", json={
        "email": "test@example.com",
        "username": "test",
        "full_name": "Test User"
    })
    assert r1.status_code == 200

    # Duplicate with a different case
    r2 = await client.post("/users", json={
        "email": "TEST@EXAMPLE.COM",
        "username": "test2",
        "full_name": "Test 2"
    })
    assert r2.status_code == 409  # citext detects duplicate


@pytest.mark.asyncio
async def test_search_with_typo_suggestions(client: AsyncClient):
    """pg_trgm: search with a typo brings suggestions."""
    # Create users
    await client.post("/users", json={
        "email": "john@example.com", "username": "john",
        "full_name": "John Smith"
    })
    await client.post("/users", json={
        "email": "jane@example.com", "username": "jane",
        "full_name": "Jane Doe"
    })

    # Search with a typo
    response = await client.get("/users/search?q=Jhon")
    body = response.json()

    # No exact match for "Jhon"
    assert len(body["results"]) > 0  # but fuzzy matches
    assert "John Smith" in body["did_you_mean"]


@pytest.mark.asyncio
async def test_uuid_format(client: AsyncClient):
    """gen_random_uuid: IDs are valid UUIDs."""
    import uuid as uuid_lib

    response = await client.post("/users", json={
        "email": "uuid@example.com", "username": "uuiduser",
        "full_name": "UUID Test"
    })
    user_id = response.json()["id"]

    # Verify UUID format
    parsed = uuid_lib.UUID(user_id)
    assert parsed.version == 4  # gen_random_uuid produces v4


@pytest.mark.asyncio
async def test_uuid_unpredictable(client: AsyncClient):
    """UUIDs are not predictable (not incremental)."""
    ids = []
    for i in range(5):
        r = await client.post("/users", json={
            "email": f"user{i}@example.com", "username": f"user{i}",
            "full_name": f"User {i}"
        })
        ids.append(r.json()["id"])

    # IDs are not sequential
    # (technically possible but astronomically unlikely)
    assert len(set(ids)) == 5  # all distinct

The final BENCHMARKS.md

# Users Refactor — Comparison Before/After

## Setup

- PostgreSQL 16
- Extensions: citext, pg_trgm, pgcrypto (built-in)
- 100,000 users seeded with realistic full_names (Faker)

## Comparison: queries by email

| Approach | Query | Plan | Time |
|----------|-------|------|------|
| Before (VARCHAR + LOWER) | `WHERE LOWER(email) = LOWER(:e)` | Index Scan on uq_users_email_lower | 12ms |
| After (CITEXT) | `WHERE email = :e` | Index Scan on uq_users_email | 11ms |

Similar performance. CITEXT is cleaner in code.

## Comparison: fuzzy search

| Approach | Query | Plan | Time |
|----------|-------|------|------|
| Before (no pg_trgm) | `WHERE full_name ILIKE '%john%'` | Seq Scan + Filter | 145ms (100k rows) |
| After (with GIN trigram) | `WHERE full_name % 'jhon'` | Bitmap Index Scan | 18ms |

8x faster with GIN trigram, plus typo tolerance.

## Comparison: PK size

| Type | Index Size (100k rows) |
|------|----------------------|
| SERIAL (BIGINT 8 bytes) | 1.5 MB |
| UUID (16 bytes) | 4.2 MB |

UUID 3x larger, but a gain in globally unique + not exposing count.

## Takeaways

1. **CITEXT** eliminates LOWER() boilerplate without significant overhead.
2. **gen_random_uuid()** trade-off: 3x size for security/distribution wins.
3. **pg_trgm** GIN index transforms fuzzy search from 145ms to 18ms.
4. **3 extensions, 0 extra dependencies** — all in PostgreSQL.

Module wrap-up

What you learned across the 8 capsules:

  1. Capsule 01: Module introduction — decision matrix-first.
  2. Capsule 02: Installation + cloud providers — operational gotchas.
  3. Capsule 03: citext vs LOWER() — decision matrix.
  4. Capsule 04: UUIDs — gen_random_uuid vs uuid-ossp.
  5. Capsule 05: hstore vs JSONB — simple rule.
  6. Capsule 06: pg_trgm advanced cases — dedup + did-you-mean.
  7. Capsule 07: Large extensions — awareness without going deep.
  8. Capsule 08: Integrating mini-project.

Your next step:

  • Apply the decisions to your real project.
  • Check the availability of extensions on your cloud provider.
  • Incremental refactor — start with citext for emails.

We start in the next module

Module 8 closes out the guide. It covers Recursive CTEs (recursive Common Table Expressions) — the SQL technique for hierarchical queries and graph traversal. And then the final capstone project: a refactor of the Blog API applying JSONB + FTS + partitioning + MVs + advisory locks + extensions + CTEs in a single codebase.

It's the close of the guide. Your portfolio ends up complete with a project that demonstrates mastery of PostgreSQL's advanced features.

Resources

  1. PostgreSQL Docs — Extensions — reference.
  2. SQLAlchemy — PostgreSQL types — all the custom types.
  3. GitHub — Awesome Postgres — curated resources.
  4. Crunchy Data Blog — analysis of extensions.

Capsule 08 of 08 — Module 7 — Advanced PostgreSQL for Backend Guide

End of module 7. Continue with module 8 (Recursive CTEs + Final Project).

← PreviousLarge extensions (mention): `pgcrypto`, `postgis`, `pgvector`Next →Module 8: Recursive CTEs + Final Project