Pattern: polymorphic data with JSONB and discriminated unions

Capsule description

The third and final canonical JSONB pattern. Your events table (or audit_log, notifications, webhook_deliveries) stores entities of the same "logical type" but with a different payload depending on a discriminator: a purchase has amount and currency; a signup has referrer and source; a view has page and duration_ms. Three options for modeling it: three separate tables (over-engineering), one table with every possible column (most of them empty), or one table with a payload JSONB validated by a discriminator.

This capsule teaches you the third option, applied with SQLAlchemy + Pydantic v2 + GIN. You'll learn why the discriminator drives the schema (not the table), how to put the GIN to work for queries by type (payload @> {"event_type": "purchase"}), how to process batches of polymorphic events without giant if-elif chains, and the complete events and audit_log pattern that shows up in real systems every day.

By the end you'll have the three canonical JSONB patterns internalized and you'll be ready for the module's project (capsule 08), where you apply everything to the Blog API refactor.

Mental model: the discriminator drives the schema, not the table

Three event types. Three ways to model them.

──────────────────────────────────────────────────────────────────────
OPTION 1: three separate tables
──────────────────────────────────────────────────────────────────────
purchases (id, user_id, amount, currency, ts)
signups (id, user_id, referrer, source, ts)
views (id, user_id, page, duration_ms, ts)

❌ A chronological listing requires a UNION ALL of three tables.
❌ Cross-type aggregations (events per user) are verbose.
❌ Adding a new type means a new table, a new model, a new endpoint.
✅ DB-level constraints per table (CHECK amount > 0).

──────────────────────────────────────────────────────────────────────
OPTION 2: one table with every column
──────────────────────────────────────────────────────────────────────
events (id, user_id, event_type, ts,
  amount, currency,         -- purchase only
  referrer, source,         -- signup only
  page, duration_ms         -- view only
)

❌ Most columns NULL.
❌ The schema grows uncontrolled with every new type.
❌ Awkward CHECK constraints ("CHECK (event_type='purchase' OR amount IS NULL)").
✅ Simple queries, one single table.

──────────────────────────────────────────────────────────────────────
OPTION 3: a table with a JSONB payload + a discriminator
──────────────────────────────────────────────────────────────────────
events (id, user_id, event_type, ts, payload JSONB)
                     ──────────         ────────────
                     "the type"         "the type's data"

✅ Chronological listing is trivial.
✅ Cross-type aggregations are simple.
✅ Adding a type: only Pydantic, no migration.
✅ GIN on payload + B-tree index on event_type → fast queries.
✅ Application-level validation with discriminated unions.
✅ Minimal DB-level constraint: event_type NOT NULL.

──────────────────────────────────────────────────────────────────────

The mental trick: the table is relational (id, user_id, event_type, ts — the common fields); the payload is polymorphic (JSONB validated in Python). That separation is the key to the pattern.

Why the discriminator is also a typed column:

event_type can live as a key inside the payload (payload.event_type) or as its own column. Recommendation: its own column (it can coexist as a key in the payload for simplicity). Reasons:

• A B-tree index on event_type is very fast for WHERE event_type = 'purchase'.
• A CHECK constraint becomes possible (CHECK (event_type IN ('purchase', 'signup', 'view'))).
• Easier to inspect from pure SQL tooling.
• If the discriminator only lives in JSONB, queries depend on the GIN; if it's a column, there's an efficient alternative for the cases where the GIN doesn't apply.

The anchor case: an events table with three event types

You're going to build the complete schema. SQLAlchemy model + Pydantic discriminated union + endpoints + queries.

SQLAlchemy model

# events_model.py
from datetime import datetime
from typing import Any, Literal

from sqlalchemy import BigInteger, CheckConstraint, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


# Discriminator as a Literal, to share with Pydantic
EventType = Literal["purchase", "signup", "view"]


class Event(Base):
    __tablename__ = "events"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    # Discriminator as a column
    event_type: Mapped[str] = mapped_column(String(32), nullable=False)
    user_id: Mapped[int | None] = mapped_column(BigInteger)
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )
    # Payload with a different shape per type
    payload: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB),
        nullable=False,
        default=dict,
    )

    __table_args__ = (
        # CHECK to limit the discriminator's valid values
        CheckConstraint(
            "event_type IN ('purchase', 'signup', 'view')",
            name="ck_events_event_type",
        ),
        # B-tree for filters by type
        Index("ix_events_event_type", "event_type"),
        # Composite index for "a user's events in chronological order"
        Index("ix_events_user_id_created_at", "user_id", "created_at"),
        # GIN on payload for containment queries
        Index("ix_events_payload_gin", "payload", postgresql_using="gin"),
    )

Schema decisions:

• event_type as a String(32) column + a CheckConstraint enumerating the values. When you add a new type, a short migration extends the CHECK.
• user_id nullable: some events are anonymous (a view without a login, for example).
• Three indexes: a B-tree for filters by type, a composite B-tree for chronological queries per user, a GIN for queries by payload content.
• MutableDict.as_mutable(JSONB): in case you ever need to mutate the payload after loading it (rare for events, but useful for audit logs that enrich the detail).

Pydantic discriminated union

# events_pydantic.py
from datetime import datetime
from typing import Annotated, Literal

from pydantic import BaseModel, Field, IPvAnyAddress


class PurchaseEvent(BaseModel):
    event_type: Literal["purchase"]
    amount: float = Field(gt=0)
    currency: str = Field(min_length=3, max_length=3, pattern=r"^[A-Z]{3}$")
    product_id: int
    quantity: int = Field(ge=1, default=1)


class SignupEvent(BaseModel):
    event_type: Literal["signup"]
    referrer: str | None = Field(default=None, max_length=200)
    source: Literal["organic", "paid", "referral", "direct"] = "organic"
    utm_campaign: str | None = Field(default=None, max_length=100)


class ViewEvent(BaseModel):
    event_type: Literal["view"]
    page: str = Field(min_length=1, max_length=500)
    duration_ms: int = Field(ge=0)
    ip_address: IPvAnyAddress | None = None


# Discriminated union: Pydantic uses event_type to pick the schema
EventPayload = Annotated[
    PurchaseEvent | SignupEvent | ViewEvent,
    Field(discriminator="event_type"),
]

How the discriminated union works (recap from capsule 04):

Pydantic looks at the input's event_type key. If it's "purchase", it validates against PurchaseEvent (requiring amount, currency, product_id). If it's "signup", against SignupEvent (accepting an optional referrer, a source with fixed values). If it's "view", against ViewEvent.

If event_type is unknown, Pydantic returns a specific error: "input type 'X' did not match any discriminator value".

Endpoint that receives polymorphic events

# events_api.py
import asyncio
from typing import Any

from fastapi import FastAPI, HTTPException, Query, status
from pydantic import BaseModel
from sqlalchemy import desc, select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


engine = create_async_engine(
    "postgresql+asyncpg://postgres:postgres@localhost:5432/demo"
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
app = FastAPI()


class CreateEventRequest(BaseModel):
    user_id: int | None = None
    payload: EventPayload  # discriminated union


@app.post("/events", status_code=status.HTTP_201_CREATED)
async def create_event(body: CreateEventRequest) -> dict[str, Any]:
    """
    FastAPI validates the body. If payload.event_type doesn't match
    any variant, 422 before entering the function.
    """
    payload_dict = body.payload.model_dump(mode="json")
    async with SessionLocal() as session:
        event = Event(
            event_type=body.payload.event_type,
            user_id=body.user_id,
            payload=payload_dict,
        )
        session.add(event)
        await session.commit()
        return {
            "id": event.id,
            "event_type": event.event_type,
            "created_at": event.created_at.isoformat(),
        }

Example requests:

# Purchase: valid
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "user_id": 42,
  "payload": {
    "event_type": "purchase",
    "amount": 99.99,
    "currency": "USD",
    "product_id": 7,
    "quantity": 2
  }
}'
# 201 Created

# View: valid, anonymous
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "payload": {
    "event_type": "view",
    "page": "/landing",
    "duration_ms": 1500
  }
}'
# 201 Created

# Incomplete purchase: amount missing
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "payload": {
    "event_type": "purchase",
    "currency": "USD",
    "product_id": 7
  }
}'
# 422 with the detail: "amount: Field required"

# Unknown discriminator
curl -X POST http://localhost:8000/events -H "Content-Type: application/json" -d '{
  "payload": {
    "event_type": "deletion",
    "user_id": 42
  }
}'
# 422 with: "Input type 'deletion' did not match any discriminator value"

Queries by discriminator: two roads to the same destination

For "every purchase in the last month," you have two equally efficient options:

Option A: filter by the event_type column

from datetime import datetime, timedelta


async def list_purchases_last_month(session) -> list[Event]:
    cutoff = datetime.utcnow() - timedelta(days=30)
    stmt = (
        select(Event)
        .where(
            Event.event_type == "purchase",
            Event.created_at >= cutoff,
        )
        .order_by(desc(Event.created_at))
    )
    return list((await session.execute(stmt)).scalars().all())

Generated SQL:

SELECT events.id, events.event_type, events.user_id, events.created_at, events.payload
FROM events
WHERE events.event_type = 'purchase'
  AND events.created_at >= '2026-04-02 00:00:00'
ORDER BY events.created_at DESC;

Indexability: it uses the B-tree index on event_type to filter, and then the one on created_at (if it exists standalone) or the composite one. Very fast.

Option B: filter by payload containment

async def list_purchases_with_filter(
    session, currency: str | None = None
) -> list[Event]:
    cutoff = datetime.utcnow() - timedelta(days=30)
    pattern = {"event_type": "purchase"}
    if currency:
        pattern["currency"] = currency
    stmt = (
        select(Event)
        .where(
            Event.payload.contains(pattern),
            Event.created_at >= cutoff,
        )
        .order_by(desc(Event.created_at))
    )
    return list((await session.execute(stmt)).scalars().all())

Generated SQL:

SELECT events.id, ...
FROM events
WHERE events.payload @> '{"event_type": "purchase", "currency": "USD"}'
  AND events.created_at >= '2026-04-02 00:00:00'
ORDER BY events.created_at DESC;

Indexability: it uses the GIN on payload. Very fast when you filter by several payload keys (@> with multiple fields is where the GIN shines).

When to use which:

• Filtering by type only: option A. A B-tree on a column is simpler and more selective.
• Filtering by type + payload fields: option B. @> with all the conditions leverages the GIN for a single indexed lookup.
• Combinations: both can coexist; the planner picks the best plan based on selectivity.

Validating the payload after reading

async def list_purchases_validated(session) -> list[PurchaseEvent]:
    stmt = select(Event).where(Event.event_type == "purchase").limit(100)
    raw_events = (await session.execute(stmt)).scalars().all()
    return [PurchaseEvent.model_validate(e.payload) for e in raw_events]

Validating on read protects you against:

• Old, malformed data (an old purchase where currency was stored as lowercase usd).
• Schema changes without a migration.
• Bugs from direct inserts that bypass the API.

If model_validate fails, you decide: skip the event, return a 500, or log it and return the "raw" event (with a warning). The policy depends on the product.

Processing polymorphic events without giant if-elif chains

The naive pattern, without a discriminator:

# Anti-pattern
def process(payload: dict):
    if payload["event_type"] == "purchase":
        amount = payload.get("amount")
        if amount is None or amount <= 0:
            raise ValueError("invalid amount")
        # ... process purchase
    elif payload["event_type"] == "signup":
        # ... process signup
    elif payload["event_type"] == "view":
        # ... process view

Manual validation, type narrowing by strings, easy to break. With discriminated unions, the pythonic pattern:

from typing import Annotated, Literal


def process_event(event: EventPayload) -> dict[str, Any]:
    """
    Clean pattern matching. Pydantic already validated.
    Automatic type narrowing: inside each branch, mypy/pyright
    knows the concrete type.
    """
    match event:
        case PurchaseEvent(amount=amount, currency=currency, product_id=pid, quantity=qty):
            return {"action": "charged", "total": amount * qty, "currency": currency, "product": pid}
        case SignupEvent(referrer=ref, source=src):
            return {"action": "user_registered", "via": src, "referrer": ref}
        case ViewEvent(page=page, duration_ms=ms):
            return {"action": "page_view", "page": page, "engagement": "high" if ms > 5000 else "low"}

Advantages:

• Pydantic validated before you got here. You don't need to check if amount is None.
• match with destructuring extracts the fields straight into typed variables.
• If you add a new type to EventPayload and forget the case in process_event, mypy/pyright warns you (in strict mode).
• No if/elif on type strings.

If your Python doesn't support match (Python <3.10), a dispatch dict works:

def process_purchase(e: PurchaseEvent) -> dict: ...
def process_signup(e: SignupEvent) -> dict: ...
def process_view(e: ViewEvent) -> dict: ...

HANDLERS = {
    PurchaseEvent: process_purchase,
    SignupEvent: process_signup,
    ViewEvent: process_view,
}


def process_event(event: EventPayload) -> dict:
    handler = HANDLERS[type(event)]
    return handler(event)

Real case: an audit log with a polymorphic detail

Another classic scenario. Your app needs to record user actions for auditing. Each action has a different detail:

• login: IP, user-agent.
• password_change: who forced it (an admin), the reason.
• role_change: previous role, new role, target user.
• delete_post: the deleted post, an optional reason.

Schema:

# audit_log.py
from datetime import datetime
from typing import Annotated, Any, Literal

from pydantic import BaseModel, Field, IPvAnyAddress
from sqlalchemy import BigInteger, CheckConstraint, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import Mapped, mapped_column


AUDIT_ACTIONS = ("login", "password_change", "role_change", "delete_post")


class AuditLog(Base):
    __tablename__ = "audit_log"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    actor_user_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    action: Mapped[str] = mapped_column(String(64), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )
    detail: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB), nullable=False, default=dict
    )

    __table_args__ = (
        CheckConstraint(
            f"action IN {AUDIT_ACTIONS!r}",
            name="ck_audit_log_action",
        ),
        Index("ix_audit_log_actor_user_id", "actor_user_id"),
        Index("ix_audit_log_action", "action"),
        Index("ix_audit_log_detail_gin", "detail", postgresql_using="gin"),
    )


class LoginDetail(BaseModel):
    action: Literal["login"]
    ip_address: IPvAnyAddress
    user_agent: str = Field(max_length=500)


class PasswordChangeDetail(BaseModel):
    action: Literal["password_change"]
    forced_by_admin_id: int | None = None
    reason: str | None = Field(default=None, max_length=500)


class RoleChangeDetail(BaseModel):
    action: Literal["role_change"]
    target_user_id: int
    old_role: str = Field(max_length=50)
    new_role: str = Field(max_length=50)


class DeletePostDetail(BaseModel):
    action: Literal["delete_post"]
    post_id: int
    reason: str | None = Field(default=None, max_length=500)


AuditDetail = Annotated[
    LoginDetail | PasswordChangeDetail | RoleChangeDetail | DeletePostDetail,
    Field(discriminator="action"),
]

Useful queries:

# Every login from an IP in the last 24h
async def logins_from_ip(session, ip: str):
    cutoff = datetime.utcnow() - timedelta(hours=24)
    stmt = select(AuditLog).where(
        AuditLog.action == "login",
        AuditLog.detail.contains({"ip_address": ip}),
        AuditLog.created_at >= cutoff,
    )
    return (await session.execute(stmt)).scalars().all()


# Every role change in the last month for a specific user
async def role_changes_for_user(session, target_user_id: int):
    cutoff = datetime.utcnow() - timedelta(days=30)
    stmt = select(AuditLog).where(
        AuditLog.action == "role_change",
        AuditLog.detail.contains({"target_user_id": target_user_id}),
        AuditLog.created_at >= cutoff,
    )
    return (await session.execute(stmt)).scalars().all()


# Every action by an actor, in chronological order
async def actor_history(session, actor_id: int, limit: int = 100):
    stmt = (
        select(AuditLog)
        .where(AuditLog.actor_user_id == actor_id)
        .order_by(AuditLog.created_at.desc())
        .limit(limit)
    )
    return (await session.execute(stmt)).scalars().all()

The first two combine a filter on action (B-tree) + a filter on the detail content (GIN). Coverage from two different indexes depending on the field. The third uses the index on actor_user_id.

Why does this matter in real work?

1. Events are ubiquitous in modern backends. In-house analytics, data warehouse integrations, audit logs, webhook deliveries, notifications. They all have the polymorphic pattern. Modeling it well once saves you from redesigning later.

2. Early validation prevents dirty data. Without discriminated unions, an endpoint that accepts a dict lets through a purchase with no amount, a signup with source: "tiktok" (when you only accept 4 values). Three months later, your revenue report has holes.

3. Pattern matching scales better than if-elif. A large team adds event types without stepping on each other. The Pydantic definitions are the "living documentation" of the expected shape. A new dev reads the discriminated union and understands the whole domain.

4. One table, cross-type queries. "How many events per user this month, by type?" is trivial: SELECT user_id, event_type, count(*) FROM events GROUP BY 1, 2. With three separate tables, it'd be a UNION + GROUP BY. With every column in one table, a query full of NULLs.

5. Adding a new type is zero-migration in production. You define RefundEvent in Pydantic, add it to EventPayload, extend the action's CHECK constraint. Deploy. No new table, no columns, no downtime.

Traps and common mistakes

Mistake 1 (conceptual): putting the discriminator only in JSONB

Symptom: without an event_type column, every query by type is WHERE payload @> {"event_type": "purchase"}. When the GIN doesn't apply (queries with computed fields, complex aggregations), you end up with a Seq Scan.

Why it happens: "the discriminator is already in the payload, why duplicate it?"

Fix: the discriminator is a column in addition to living in the payload. You get a better B-tree for simple filters, optionally a CHECK constraint, more efficient indexes. The duplication is trivial (on every insert, you copy a short string) and it's worth the cost.

Mistake 2 (conceptual): not validating the payload on read

Symptom: you read an old event that was stored with a schema that has since changed, you try to use it, and it crashes.

Why it happens: you assume the database is always consistent with the current schema.

Fix: validate with the discriminated union on read (at least for batch processing). If the event doesn't validate, you decide: skip it with a log, mark it for migration, or fail.

Mistake 3 (practical): forgetting mode="json" with IPvAnyAddress and datetime

Symptom: when persisting the validated payload, an error like Object of type IPv4Address is not JSON serializable.

Why it happens: IPvAnyAddress and datetime are Python types, not JSON.

Fix: always model.model_dump(mode="json") before assigning to the JSONB.

Mistake 4 (conceptual): a discriminated union without Literal in each variant

Symptom: Pydantic tries to validate against each variant sequentially and returns confusing combined errors.

Why it happens: you declared event_type: str instead of event_type: Literal["purchase"] in one of the variants.

Fix:

class PurchaseEvent(BaseModel):
    event_type: Literal["purchase"]   # NOT str

The discriminator needs the specific Literal in each variant to do the matching.

Mistake 5 (conceptual): abusive polymorphism

Symptom: your EventPayload has 30 variants. Maintaining them becomes unsustainable.

Why it happens: every event in the system ends up in the same table.

Fix: split into tables by domain. analytics_events (purchase, signup, view) separate from audit_log (login, role_change, etc.) separate from notifications (email_sent, push_sent). Each table has its own manageable set of variants.

Mistake 6 (practical): a GIN with no strategy for heavy writes

Symptom: your events table receives thousands of inserts/second. The GIN slows it down.

Why it happens: GIN has a write cost (module 1, capsule 05). With write-heavy workloads, that cost matters.

Fix:

• WITH (fastupdate = on) (the default) accumulates updates in a pending list, making inserts cheaper.
• Consider jsonb_path_ops if you only filter with @>. A smaller index, faster writes.
• If the volume scales, module 4 (partitioning) applies: partition events by month and move the GIN to the partition.

Exercises

Exercise 1: a discriminated union for notifications

Your app sends notifications through three channels: email, push, sms. Each one has a different shape:

• email: to, subject, body_html, optional cc and bcc.
• push: device_token, title, body, optional badge_count.
• sms: phone, body (max 160 chars), optional media_url.

Define the discriminated union, the SQLAlchemy model, and a POST /notifications endpoint that receives and persists them.

from typing import Annotated, Any, Literal

from fastapi import FastAPI, status
from pydantic import BaseModel, EmailStr, Field, HttpUrl
from sqlalchemy import BigInteger, CheckConstraint, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


CHANNELS = ("email", "push", "sms")


class Notification(Base):
    __tablename__ = "notifications"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    channel: Mapped[str] = mapped_column(String(16), nullable=False)
    user_id: Mapped[int | None] = mapped_column(BigInteger)
    sent_at: Mapped[datetime] = mapped_column(
        server_default=func.now(), nullable=False
    )
    payload: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB), nullable=False
    )

    __table_args__ = (
        CheckConstraint(f"channel IN {CHANNELS!r}", name="ck_notifications_channel"),
        Index("ix_notifications_channel", "channel"),
        Index("ix_notifications_user_id", "user_id"),
        Index("ix_notifications_payload_gin", "payload", postgresql_using="gin"),
    )


class EmailPayload(BaseModel):
    channel: Literal["email"]
    to: EmailStr
    subject: str = Field(min_length=1, max_length=200)
    body_html: str = Field(min_length=1)
    cc: list[EmailStr] = Field(default_factory=list)
    bcc: list[EmailStr] = Field(default_factory=list)


class PushPayload(BaseModel):
    channel: Literal["push"]
    device_token: str = Field(min_length=1, max_length=500)
    title: str = Field(min_length=1, max_length=100)
    body: str = Field(min_length=1, max_length=500)
    badge_count: int | None = Field(default=None, ge=0)


class SmsPayload(BaseModel):
    channel: Literal["sms"]
    phone: str = Field(pattern=r"^\+\d{8,15}$")
    body: str = Field(min_length=1, max_length=160)
    media_url: HttpUrl | None = None


NotificationPayload = Annotated[
    EmailPayload | PushPayload | SmsPayload,
    Field(discriminator="channel"),
]


class CreateNotificationRequest(BaseModel):
    user_id: int | None = None
    payload: NotificationPayload


@app.post("/notifications", status_code=status.HTTP_201_CREATED)
async def create_notification(body: CreateNotificationRequest) -> dict[str, Any]:
    async with SessionLocal() as session:
        notif = Notification(
            channel=body.payload.channel,
            user_id=body.user_id,
            payload=body.payload.model_dump(mode="json"),
        )
        session.add(notif)
        await session.commit()
        return {"id": notif.id, "channel": notif.channel}

Exercise 2: an indexed query by channel and payload field

Write the query that returns the last 50 email notifications whose to is user@example.com. Which indexes does it use?

async def recent_emails_to(session, email: str) -> list[Notification]:
    stmt = (
        select(Notification)
        .where(
            Notification.channel == "email",
            Notification.payload.contains({"to": email}),
        )
        .order_by(Notification.sent_at.desc())
        .limit(50)
    )
    return list((await session.execute(stmt)).scalars().all())

SELECT notifications.id, notifications.channel, notifications.user_id,
       notifications.sent_at, notifications.payload
FROM notifications
WHERE notifications.channel = 'email'
  AND notifications.payload @> '{"to": "user@example.com"}'
ORDER BY notifications.sent_at DESC
LIMIT 50;

EXPLAIN ANALYZE
SELECT * FROM notifications
WHERE channel = 'email'
  AND payload @> '{"to": "user@example.com"}'
ORDER BY sent_at DESC LIMIT 50;
-- Expected:
-- Limit
--   ->  Sort
--         ->  Bitmap Heap Scan on notifications
--               Recheck Cond: ((channel = 'email') AND (payload @> '...'))
--               ->  BitmapAnd
--                     ->  Bitmap Index Scan on ix_notifications_channel
--                     ->  Bitmap Index Scan on ix_notifications_payload_gin

Exercise 3: processing with pattern matching

Implement a dispatch_notification(payload: NotificationPayload) -> str function that dispatches the send to the right service, using match. For each case, return the string "sent X via Y" (where X is something identifying and Y is the channel). No real send logic — just the typed dispatch.

def dispatch_notification(payload: NotificationPayload) -> str:
    """
    Dispatches the notification to the right service.
    Uses pattern matching for typed destructuring.
    """
    match payload:
        case EmailPayload(to=to, subject=subject):
            # Here you'd have: await email_service.send(to, subject, payload.body_html, ...)
            return f"sent email to {to} (subject: {subject!r})"

        case PushPayload(device_token=token, title=title):
            # Here you'd have: await push_service.send(token, title, payload.body, payload.badge_count)
            return f"sent push to device {token[:8]}... (title: {title!r})"

        case SmsPayload(phone=phone, body=body):
            # Here you'd have: await sms_service.send(phone, body, payload.media_url)
            return f"sent sms to {phone} ({len(body)} chars)"

import pytest


def test_dispatch_email():
    payload = EmailPayload(
        channel="email",
        to="user@example.com",
        subject="Hello",
        body_html="<p>Test</p>",
    )
    result = dispatch_notification(payload)
    assert "sent email to user@example.com" in result


def test_dispatch_push():
    payload = PushPayload(
        channel="push",
        device_token="abcdefghijklmnop",
        title="Alert",
        body="Something happened",
    )
    result = dispatch_notification(payload)
    assert result.startswith("sent push to device abcdefgh")

Exercise 4: adding a new event type without a migration

Your product adds a new event type: refund. It has original_purchase_id, amount, currency, reason. Document the exact steps to integrate it. What migrations do you need?

class RefundEvent(BaseModel):
    event_type: Literal["refund"]
    original_purchase_id: int
    amount: float = Field(gt=0)
    currency: str = Field(min_length=3, max_length=3, pattern=r"^[A-Z]{3}$")
    reason: str = Field(max_length=500)

EventPayload = Annotated[
    PurchaseEvent | SignupEvent | ViewEvent | RefundEvent,
    Field(discriminator="event_type"),
]

# alembic/versions/xxxx_add_refund_event_type.py
import sqlalchemy as sa
from alembic import op


def upgrade() -> None:
    op.drop_constraint("ck_events_event_type", "events", type_="check")
    op.create_check_constraint(
        "ck_events_event_type",
        "events",
        "event_type IN ('purchase', 'signup', 'view', 'refund')",
    )


def downgrade() -> None:
    op.drop_constraint("ck_events_event_type", "events", type_="check")
    op.create_check_constraint(
        "ck_events_event_type",
        "events",
        "event_type IN ('purchase', 'signup', 'view')",
    )

def process_event(event: EventPayload) -> dict[str, Any]:
    match event:
        ...
        case RefundEvent(original_purchase_id=pid, amount=amt, currency=cur):
            return {"action": "refunded", "of": pid, "amount": amt, "currency": cur}

Exercise 5: spot and refactor a badly typed payload

The following code exists in production. Spot the problems and propose a refactor.

class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    payload: Mapped[dict] = mapped_column(JSONB, nullable=False)


@app.post("/events")
async def create_event(payload: dict) -> dict:
    async with SessionLocal() as session:
        # No validation, no discriminator
        event = Event(payload=payload)
        session.add(event)
        await session.commit()
        return {"id": event.id}


@app.get("/events/purchases")
async def list_purchases() -> list[dict]:
    async with SessionLocal() as session:
        result = await session.execute(
            select(Event).where(
                Event.payload["event_type"].astext == "purchase"
            )
        )
        return [e.payload for e in result.scalars().all()]

from typing import Annotated, Any, Literal
from pydantic import BaseModel, Field
from sqlalchemy import CheckConstraint, Index, String


class Event(Base):
    __tablename__ = "events"
    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    event_type: Mapped[str] = mapped_column(String(32), nullable=False)
    payload: Mapped[dict[str, Any]] = mapped_column(
        MutableDict.as_mutable(JSONB), nullable=False
    )

    __table_args__ = (
        CheckConstraint(
            "event_type IN ('purchase', 'signup', 'view')",
            name="ck_events_event_type",
        ),
        Index("ix_events_event_type", "event_type"),
        Index("ix_events_payload_gin", "payload", postgresql_using="gin"),
    )


class PurchaseEvent(BaseModel):
    event_type: Literal["purchase"]
    amount: float = Field(gt=0)
    currency: str = Field(min_length=3, max_length=3)
    user_id: int


class SignupEvent(BaseModel):
    event_type: Literal["signup"]
    user_id: int
    referrer: str | None = None


class ViewEvent(BaseModel):
    event_type: Literal["view"]
    page: str
    user_id: int | None = None


EventPayload = Annotated[
    PurchaseEvent | SignupEvent | ViewEvent,
    Field(discriminator="event_type"),
]


@app.post("/events", status_code=201)
async def create_event(payload: EventPayload) -> dict[str, int]:
    async with SessionLocal() as session:
        event = Event(
            event_type=payload.event_type,
            payload=payload.model_dump(mode="json"),
        )
        session.add(event)
        await session.commit()
        return {"id": event.id}


@app.get("/events/purchases", response_model=list[PurchaseEvent])
async def list_purchases() -> list[PurchaseEvent]:
    async with SessionLocal() as session:
        result = await session.execute(
            select(Event)
            .where(Event.event_type == "purchase")  # now a B-tree index
            .order_by(Event.id.desc())
            .limit(100)
        )
        events = result.scalars().all()
        return [PurchaseEvent.model_validate(e.payload) for e in events]

Summary and next step

In this capsule you learned the third canonical JSONB pattern:

• Polymorphic data: the same table, payloads with different shapes depending on a discriminator.
• The discriminator as a column (not only in JSONB) for B-tree indexes and CHECK constraints.
• Pydantic discriminated unions validate the type-specific shape, give you type narrowing, and scale better than if-elif.
• Two roads to the same destination for queries by type: a B-tree on the column, or a GIN on the payload with @>. They combine.
• Pattern matching processes polymorphic events without if-elif, with typed destructuring.
• Adding a new type is Pydantic + extending the CHECK. Zero column migration, zero downtime.
• The audit log is the flip side of the same pattern: action + detail JSONB, with a discriminated union.

Before moving on you should be able to:

• Design a table with a payload JSONB + a discriminator column + the right indexes.
• Define a Pydantic discriminated union with three or more variants.
• Write an endpoint that receives validated polymorphic payloads.
• Query by type while leveraging the right indexes.
• Process events with typed pattern matching.

Next capsule — Project: extending the Blog API with JSONB. You'll apply everything you learned in this module. The Blog API you built in guide #8 has posts with several loose optional fields (seo_title VARCHAR, seo_description VARCHAR, etc.). You'll refactor it by adding a metadata JSONB column, validating it with Pydantic, indexing it with a GIN, and migrating the existing data with an idempotent Alembic script. The deliverable is PR-style, with tests and benchmarks. It's the first step of module 8's integrative project.

Resources

1. Pydantic v2 — Discriminated Unions — the complete official reference.
2. Python 3.10+ — Structural Pattern Matching — the PEP 636 tutorial on the match statement.
3. PostgreSQL 16 — CHECK Constraints — to limit the discriminator's values at the DB level.
4. PostgreSQL 16 — JSONB Containment — the semantics of @> that the GIN accelerates.
5. SQLAlchemy 2.0 — func and query composition — for combined queries.
6. Martin Fowler — Domain Event — the conceptual reference on why events are ubiquitous in modern backends.
7. Bruce Momjian — Postgres JSON Tricks — JSONB usage patterns from a core team member.

Module 2 — Advanced PostgreSQL for Backend Guide

Next capsule: The module's project — refactoring the Blog API by adding posts.metadata with an Alembic migration, Pydantic validation, and GIN indexing.