Thursday, May 8, 2025

Pydantic v2: Discriminated Unions + Fallbacks - Graceful Webhook Parsing Example

TL;DR: If you're handling webhook events from Mux (or any webhook-heavy API) and only care about a subset of event types, use an Annotated Union with discriminator="type" wrapped in an Annotated Union with a generic fallback and union_mode="left_to_right" to cleanly support known types and fall back to a generic model for unknown ones.


How I Got Here

I started this project with a simple goal: handle webhook events from Mux.com. Their webhook payloads include a type field and a data object, and I only cared about a few specific types like video.asset.created or video.asset.ready. Unlike the rest of the Mux API, they do not provide an SDK with types for webhook responses.

I've been wanting to use FastAPI and this was a perfect spot. I found Pydantic v2's discriminated union support and was excited to see something to leverage Python typing that felt more 🦀-like 🧡. It seemed perfect: define the type field, use Literal[...], and Pydantic would match things for me super efficiently.

Until it didn't.

The catch: Pydantic's discriminator requires exact matches to a Literal[...]. If an unknown type comes in (which is common with evolving APIs), Pydantic throws a validation error. That's not great for a webhook handler that should be resilient and fault-tolerant.

I wanted something more forgiving. Something like Rust's match on tagged enums, but with a wildcard fallback.

In Pydantic v1, the default union_mode was "left_to_right", which tried the models one after another, like a long if-elif chain. In Pydantic v2, there are now three union_modes: "left_to_right", "smart_mode", and "discriminator". The default is now "smart_mode", which continues after a match looking for “better matches.” I wanted a way to use my discriminated union if possible, but also have a catch-all.

The trick that makes this work is nesting: you define your discriminated union inside a broader union that includes your generic fallback type. The outer union uses union_mode="left_to_right", so it tries the discriminated logic first, and if no match is found, it falls back.

The Core Pattern

We combine two ideas:

  • A discriminator="type" union of known types we care about
  • A fallback union using Field(union_mode="left_to_right") so unknown types resolve to a generic handler

This lets us cleanly parse webhook data with type-safe models for the cases we care about and a resilient fallback when we don't.

Simplified Example

from typing import Annotated, Union, Literal
from pydantic import BaseModel, Field

class GenericWebhookEvent(BaseModel):
    type: str
    data: dict  # fallback catch-all

class VideoAssetCreated(GenericWebhookEvent):
    type: Literal["video.asset.created"]
    data: dict  # your typed payload here

# This inner union uses discriminator on the "type" field, so each member must have a Literal[...] match
KnownWebhookEvents = Annotated[
    Union[
        VideoAssetCreated,
    ],
    Field(discriminator="type")
]

# This outer union attempts KnownWebhookEvents first, then falls back to GenericWebhookEvent
WebhookEventUnion = Annotated[
    Union[
        KnownWebhookEvents,
        GenericWebhookEvent,
    ],
    Field(union_mode="left_to_right")
]

Now parsing behaves like this:

WebhookEventUnion.model_validate({
    "type": "video.asset.created",
    "data": {...}
})
# => VideoAssetCreated

WebhookEventUnion.model_validate({
    "type": "video.asset.unknown",
    "data": {...}
})
# => GenericWebhookEvent

Real-World Version

In production, you can organize your types like this:

video_asset.py

from typing import Literal, Annotated
from pydantic import Field
from .generic_webhook import GenericWebhookEvent, GenericWebhookData

class VideoAssetCreated(GenericWebhookEvent):
    type: Literal["video.asset.created"]
    data: VideoAssetCreatedData

class VideoAssetCreatedData(GenericWebhookData):
    id: str
    duration: float

VideoAssetUnion = Annotated[
    Union[
        VideoAssetCreated,
        # Add other known video.asset types here
    ],
    Field(discriminator="type")
]

generic_webhook.py

from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

class GenericWebhookData(BaseModel):
    id: str
    status: Optional[str] = None
    created_at: Optional[datetime] = None

class GenericWebhookEvent(BaseModel):
    type: str
    id: str
    data: GenericWebhookData
    created_at: Optional[datetime] = None

__init__.py

from typing import Annotated, Union
from pydantic import Field

from .video_asset import VideoAssetUnion
from .generic_webhook import GenericWebhookEvent

# Tries the VideoAssetUnion first via discriminator logic, then falls back to generic
WebhookEventUnion = Annotated[
    Union[
        VideoAssetUnion,
        GenericWebhookEvent,
    ],
    Field(union_mode="left_to_right")
]

Why This Pattern Works

  • ✅ Strong typing for the payloads you care about
  • ✅ Unknown events don’t crash your webhook parser
  • ✅ Clean and extensible: just add new types to the top-level union
  • ✅ Better than writing massive if-else chains

When To Use It

This is ideal when:

  • You're consuming webhook payloads
  • You only care about a subset of known types
  • You want to extend coverage gradually as needed

Bonus Tip

Use .model_validate_json(...) if you're pulling webhooks directly from an HTTP request body.


Summary

Pydantic v2's discriminated union support is great, but it's picky. If you're working with a webhook API and want to use discriminator, be prepared for unknown types to raise errors.

Fix that with union_mode="left_to_right". It gives you the best of both worlds: expressive type-safe unions with a graceful fallback.

This left-to-right pattern using Annotated[Union[..., fallback], Field(...)] is my new favorite way to parse webhook payloads.


✍️ Feel free to fork the gist or adapt this pattern to other APIs like Stripe, GitHub, or Slack!

Questions or improvements? Drop them below or @ me on GitHub.

No comments: