Message Validation

~12 min · protocol, pydantic, validation

Level 0Poller

0 XP0/60 lessons0/10 achievements

0/120 XP to next level120 XP to go0% complete

Never trust client input

Every incoming WebSocket message is hostile until proven otherwise. Validate the envelope, the type, every field, the lengths. Pydantic gives you structured validation with one decorator and one schema, with errors you can return as {type: 'error', code: 'validation_error', message: ...}.

Tagged union with Literal

Define a Pydantic model per message type and use Field(discriminator='type') to dispatch. The framework picks the right model based on the type field, validates the rest, and returns a typed object. Bad messages reject before any handler runs.

Code

Pydantic discriminated union·python

from pydantic import BaseModel, Field, ValidationError, field_validator
from typing import Literal, Annotated, Union, Optional

class ChatPayload(BaseModel):
    room: str
    text: str
    reply_to: Optional[str] = None

    @field_validator('text')
    @classmethod
    def text_ok(cls, v: str):
        v = v.strip()
        if not v:
            raise ValueError('empty')
        if len(v) > 5_000:
            raise ValueError('too long')
        return v

class ChatMessage(BaseModel):
    type: Literal['chat.message']
    data: ChatPayload

class JoinPayload(BaseModel):
    room: str

class RoomJoin(BaseModel):
    type: Literal['room.join']
    data: JoinPayload

InboundMessage = Annotated[
    Union[ChatMessage, RoomJoin],
    Field(discriminator='type'),
]

# In your handler
async def handle(ws, raw: dict):
    try:
        msg = pydantic.TypeAdapter(InboundMessage).validate_python(raw)
    except ValidationError as e:
        await ws.send_json({
            'type': 'error',
            'code': 'validation_error',
            'message': e.errors()[0]['msg'],
        })
        return
    if isinstance(msg, ChatMessage):
        await on_chat(ws, msg.data)
    elif isinstance(msg, RoomJoin):
        await on_join(ws, msg.data)

External links

Pydantic: Discriminated unions

Exercise

Add Pydantic validation for three message types. Send a malformed chat.message (empty text); a bad type field ('chat.unknown'); a missing field. Confirm each produces a structured error response that the client can react to without crashing.