~12 min · protocol, pydantic, validation
들어오는 모든 WebSocket message 가 입증 전엔 적대적. envelope, type, 모든 field, 길이 다 validate. Pydantic 이 한 decorator + 한 schema 로 structured validation, error 는 {type: 'error', code: 'validation_error', message: ...} 로 리턴 가능.
Message type 별 Pydantic 모델 정의 후 Field(discriminator='type') 로 dispatch. 프레임워크가 type field 기반으로 옳은 모델 픽, 나머지 validate, typed 객체 리턴. 나쁜 message 는 handler 돌기 전 reject.
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)chat.message (빈 text), 잘못된 type field ('chat.unknown'), 빠진 field 보내. 각각이 클라가 crash 없이 react 가능한 structured error 응답 만드는지 확인.아직 댓글이 없어요. 첫 댓글을 남겨보세요.