Объявить WebSocket API#

Три формы, три результата.

WebSocket runtime отделён от HTTP-клиента, но использует тот же подход к операциям, моделям, middleware и crypto.

uv add "eazy-sdk-core[websocket]"

# examples/docs/websocket.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Notify(WsSend):
    __ws__ = Ws.send("notify")
    value: int


@dataclass(frozen=True, slots=True, kw_only=True)
class Lookup(WsCall[str]):
    __ws__ = Ws.call("lookup")
    name: str


@dataclass(frozen=True, slots=True, kw_only=True)
class Prices(WsSubscribe[PriceEvent]):
    __ws__ = Ws.subscribe("prices")
    symbol: str


class MarketApi(AsyncWsApi):
    notify = op(Notify)
    lookup = op(Lookup)
    prices = op(Prices)


def main() -> None:
    for operation in (Notify, Lookup, Prices):
        print(f"{operation.__ws__.kind.value}: {operation.__ws__.discriminator}")

WsSend отправляет сообщение без ответа. WsCall[T] ждёт один типизированный ответ, а WsSubscribe[T] возвращает поток событий. Все поля класса входят в payload.

Файл целиком
"""Declare send, call and subscription WebSocket operations."""

from dataclasses import dataclass

from pydantic import BaseModel

from eazy_sdk import op
from eazy_sdk.websocket import AsyncWsApi, Ws, WsCall, WsSend, WsSubscribe


class PriceEvent(BaseModel):
    value: int


# region docs: websocket
# examples/docs/websocket.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Notify(WsSend):
    __ws__ = Ws.send("notify")
    value: int


@dataclass(frozen=True, slots=True, kw_only=True)
class Lookup(WsCall[str]):
    __ws__ = Ws.call("lookup")
    name: str


@dataclass(frozen=True, slots=True, kw_only=True)
class Prices(WsSubscribe[PriceEvent]):
    __ws__ = Ws.subscribe("prices")
    symbol: str


class MarketApi(AsyncWsApi):
    notify = op(Notify)
    lookup = op(Lookup)
    prices = op(Prices)


def main() -> None:
    for operation in (Notify, Lookup, Prices):
        print(f"{operation.__ws__.kind.value}: {operation.__ws__.discriminator}")
# endregion docs: websocket


if __name__ == "__main__":
    main()

send: notify
call: lookup
subscribe: prices

Выбрать протокол и жизненный цикл#

JsonEventProtocol подходит для событий с discriminator. GraphqlTransportWsProtocol реализует graphql-transport-ws, включая connection_init, connection_ack, next, error и complete.

Credential

Когда читается

StaticUpgradeAuth

При создании Zapros-клиента для HTTP upgrade

ProtocolAuth

Для каждого нового поколения соединения

DynamicPerMessageAuth

Перед каждым прикладным сообщением

Повтор send или call требует явной политики, потому что доставка после разрыва может быть неизвестна. Для subscription отдельно задаётся resubscribe, способ восстановления позиции, ограниченная очередь и policy overflow, чтобы медленный потребитель не накапливал события без границы.

Сгенерировать из AsyncAPI#

uv add "eazy-sdk-core[asyncapi,yaml]"
eazy-sdk-asyncapi asyncapi.yaml generated --package-name market_stream

Генератор читает AsyncAPI 3.0 для ws и wss, создаёт модели, протокол и тонкий AsyncWsApi. Приложение по-прежнему владеет клиентом, reconnect policy и профилями шифрования.

Дальше#