WebSocketRoute

page.route_web_socket() または browser_context.route_web_socket() で WebSocket ルートが設定されるたびに、WebSocketRoute オブジェクトは実際のサーバーのように WebSocket を処理できます。

モック

デフォルトでは、ルーティングされた WebSocket はサーバーに接続しません。この方法で、WebSocket を介した通信全体をモックできます。以下は、"request" に対して "response" で応答する例です。

同期

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    ws.send("response")

page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

非同期

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    ws.send("response")

await page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

WebSocket ルートハンドラー内で web_socket_route.connect_to_server を呼び出さないため、Playwright は WebSocket がモックされると想定し、ページ内で WebSocket を自動的に開きます。

JSON メッセージを処理する別の例です。

同期

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  json_message = json.loads(message)
  if json_message["request"] == "question":
    ws.send(json.dumps({ "response": "answer" }))

page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

非同期

def message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  json_message = json.loads(message)
  if json_message["request"] == "question":
    ws.send(json.dumps({ "response": "answer" }))

await page.route_web_socket("wss://example.com/ws", lambda ws: ws.on_message(
    lambda message: message_handler(ws, message)
))

インターセプト

または、実際のサーバーに接続しつつ、メッセージを途中で傍受して変更またはブロックすることもできます。web_socket_route.connect_to_server を呼び出すと、サーバー側の WebSocketRoute インスタンスが返され、それにメッセージを送信したり、受信メッセージを処理したりできます。

以下は、ページからサーバーに送信される一部のメッセージを変更する例です。サーバーからページに送信されるメッセージは、デフォルトの転送に依存してそのまま残されます。

同期

def message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    server.send("request2")
  else:
    server.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: message_handler(server, message))

page.route_web_socket("/ws", handler)

非同期

def message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message == "request":
    server.send("request2")
  else:
    server.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: message_handler(server, message))

await page.route_web_socket("/ws", handler)

サーバーに接続すると、デフォルトですべての **メッセージは** ページとサーバー間で **転送されます**。

ただし、元のルートで web_socket_route.on_message() を呼び出すと、ページからサーバーへのメッセージは転送されなくなり、代わりにハンドラーによって処理される必要があります。

同様に、サーバー側の WebSocket で web_socket_route.on_message() を呼び出すと、サーバーからページへのメッセージの転送が停止され、ハンドラーがそれらを処理する必要があります。

次の例では、両方向で一部のメッセージをブロックします。両方向で web_socket_route.on_message() を呼び出すため、自動転送はまったく行われません。

同期

def ws_message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-page":
    server.send(message)

def server_message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-server":
    ws.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: ws_message_handler(server, message))
  server.on_message(lambda message: server_message_handler(ws, message))

page.route_web_socket("/ws", handler)

非同期

def ws_message_handler(server: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-page":
    server.send(message)

def server_message_handler(ws: WebSocketRoute, message: Union[str, bytes]):
  if message != "blocked-from-the-server":
    ws.send(message)

def handler(ws: WebSocketRoute):
  server = ws.connect_to_server()
  ws.on_message(lambda message: ws_message_handler(server, message))
  server.on_message(lambda message: server_message_handler(ws, message))

await page.route_web_socket("/ws", handler)

---

メソッド

close

追加日: v1.48 webSocketRoute.close

WebSocket 接続の片側を閉じます。

使用法

web_socket_route.close()
web_socket_route.close(**kwargs)

引数

• code int (任意)#

  オプションの クローズコード。

• reason str (任意)#

  オプションの クローズ理由。

戻り値

• NoneType#

---

on_close

追加日: v1.48 webSocketRoute.on_close

WebSocket.close を処理できます。

デフォルトでは、ページまたはサーバーのいずれかで接続の一方を閉じると、もう一方も閉じられます。ただし、web_socket_route.on_close() ハンドラーが設定されている場合、閉鎖のデフォルト転送は無効になり、ハンドラーがそれを処理する必要があります。

使用法

web_socket_route.on_close(handler)

引数

• handler Callable[int | [undefined]]:Promise[Any] | Any#

  WebSocket のクローズを処理する関数です。オプションの クローズコード とオプションの クローズ理由 を受け取ります。

---

on_message

追加日: v1.48 webSocketRoute.on_message

このメソッドは、WebSocket によって、ページから、またはサーバーから送信されたメッセージを処理することを可能にします。

元の WebSocket ルートで呼び出されると、このメソッドはページから送信されたメッセージを処理します。web_socket_route.send() で応答するか、web_socket_route.connect_to_server で返されたサーバー側接続に転送するか、または別の方法で処理できます。

このメソッドが呼び出されると、メッセージはサーバーまたはページに自動的に転送されません。web_socket_route.send() を呼び出して手動で転送する必要があります。詳細については、冒頭の例を参照してください。

このメソッドを再度呼び出すと、ハンドラーが新しいものに上書きされます。

使用法

web_socket_route.on_message(handler)

引数

• handler Callable[str]:Promise[Any] | Any#

  メッセージを処理する関数です。

---

send

追加日: v1.48 webSocketRoute.send

WebSocket にメッセージを送信します。元の WebSocket で呼び出された場合、メッセージはページに送信されます。web_socket_route.connect_to_server の結果で呼び出された場合、メッセージはサーバーに送信されます。詳細については、冒頭の例を参照してください。

使用法

web_socket_route.send(message)

引数

• message str | bytes#

  送信するメッセージ。

---

プロパティ

connect_to_server

追加日: v1.48 webSocketRoute.connect_to_server

デフォルトでは、ルーティングされた WebSocket はサーバーに接続しないため、WebSocket 通信全体をモックできます。このメソッドは実際の WebSocket サーバーに接続し、サーバー側の WebSocketRoute インスタンスを返して、サーバーとのメッセージの送受信を可能にします。

サーバーに接続すると

• サーバーから受信したメッセージは、サーバー側の WebSocketRoute で web_socket_route.on_message() が呼び出されない限り、ページの WebSocket に自動的に転送されます。
• ページでの WebSocket.send() 呼び出しによって送信されたメッセージは、元の WebSocketRoute で web_socket_route.on_message() が呼び出されない限り、サーバーに自動的に転送されます。

詳細については、冒頭の例を参照してください。

使用法

web_socket_route.connect_to_server

戻り値

• WebSocketRoute#

---

url

追加日: v1.48 webSocketRoute.url

ページで作成された WebSocket の URL。

使用法

web_socket_route.url

戻り値

• str#