Исключение WebSocketRequestValidationError

Исключение WebSocketRequestValidationError вызывается фреймворком FastAPI автоматически, когда входящие данные WebSocket-соединения не проходят валидацию Pydantic. Оно наследуется от встроенного класса RequestValidationError из библиотеки pydantic. Исключение содержит список ошибок, доступных через атрибут errors, и передаётся в обработчик, зарегистрированный через декоратор websocket. В отличие от HTTP-запросов, это исключение нельзя перехватить с помощью @app.exception_handler - оно возникает внутри цикла обработки WebSocket-сообщения до вызова пользовательской функции.

Синтаксис

from fastapi.exceptions import WebSocketRequestValidationError

raise WebSocketRequestValidationError(errors)

Пример

Обычно исключение возникает автоматически при получении данных через receive_json и несоответствии их Pydantic-модели. Давайте рассмотрим базовый WebSocket-обработчик, который принимает JSON и валидирует его через модель:

from fastapi import FastAPI, WebSocket
from pydantic import BaseModel

app = FastAPI()

class Message(BaseModel):
	text: str
	count: int

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
	await websocket.accept()
	data = await websocket.receive_json()
	message = Message(**data)
	await websocket.send_json({'text': message.text})
	await websocket.close()

При отправке клиентом JSON вида {"text": "hello", "count": "abc"} возникает исключение WebSocketRequestValidationError, потому что поле count ожидает целое число, а получает строку. Соединение при этом закрывается.

Пример

Чтобы перехватить ошибку валидации и отправить клиенту понятное сообщение, нужно использовать конструкцию try/except внутри WebSocket-обработчика:

from fastapi import FastAPI, WebSocket
from fastapi.exceptions import WebSocketRequestValidationError
from pydantic import BaseModel, ValidationError

app = FastAPI()

class Message(BaseModel):
	text: str
	count: int

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
	await websocket.accept()
	try:
		data = await websocket.receive_json()
		message = Message(**data)
		await websocket.send_json({'text': message.text})
	except ValidationError as exc:
		await websocket.send_json({'errors': exc.errors()})
	await websocket.close()

Результат выполнения кода при неверных данных:

{"errors": [{"type": "int_parsing", "loc": ["count"], "msg": "Input should be a valid integer, unable to parse string as an integer", "input": "abc"}]}

Пример

Также можно вручную поднять исключение WebSocketRequestValidationError и передать в него список ошибок, полученный от Pydantic:

from fastapi import FastAPI, WebSocket
from fastapi.exceptions import WebSocketRequestValidationError
from pydantic import BaseModel, ValidationError

app = FastAPI()

class Message(BaseModel):
	text: str
	count: int

@app.websocket('/ws')
async def websocket_endpoint(websocket: WebSocket):
	await websocket.accept()
	data = await websocket.receive_json()
	try:
		message = Message(**data)
	except ValidationError as exc:
		raise WebSocketRequestValidationError(exc.errors())
	await websocket.send_json({'text': message.text})
	await websocket.close()

Такая конструкция позволяет единообразно обрабатывать ошибки валидации и логировать их на стороне сервера через middleware или глобальный обработчик.

Смотрите также

исключение RequestValidationError,
которое возникает при ошибках валидации HTTP-запроса
исключение WebSocketException,
которое используется для закрытия WebSocket с кодом ошибки
исключение WebSocketDisconnect,
которое возникает при разрыве WebSocket-соединения клиентом
атрибут code,
который задаёт код закрытия WebSocket-соединения