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_mode
s: "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:
Post a Comment