# WebSockets { #websockets }

你可以在 **FastAPI** 中使用 [WebSockets](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API)。

## 安裝 `websockets` { #install-websockets }

請先建立[虛擬環境](../virtual-environments.md)、啟用它，然後安裝 `websockets`（一個讓你更容易使用「WebSocket」通訊協定的 Python 套件）：

<div class="termy">

```console
$ pip install websockets

---> 100%
```

</div>

## WebSockets 用戶端 { #websockets-client }

# WebSockets { #websockets }

您可以在 **FastAPI** 中使用 [WebSockets](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API)。

## 安装 `websockets` { #install-websockets }

请确保您创建一个[虚拟环境](../virtual-environments.md)、激活它，并安装 `websockets`（一个让使用“WebSocket”协议更容易的 Python 库）：

<div class="termy">

```console
$ pip install websockets

---> 100%
```

</div>

## WebSockets 客户端 { #websockets-client }

        .Builder()
        .webSocketUpgrade(serverListener)
        .build(),
    )
    val webSocket: WebSocket = newWebSocket()
    clientListener.assertOpen()
    val server = serverListener.assertOpen()
    webSocket.send("Hello, WebSockets!")
    serverListener.assertTextMessage("Hello, WebSockets!")
    closeWebSockets(webSocket, server)
  }

  @Test
  fun binaryMessage() {
    webServer.enqueue(

# WebSockets { #websockets }

**FastAPI**で[WebSockets](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API)が使用できます。

## `websockets`のインストール { #install-websockets }

[仮想環境](../virtual-environments.md)を作成し、それを有効化してから、「WebSocket」プロトコルを簡単に使えるようにするPythonライブラリの`websockets`をインストールしてください。

<div class="termy">

```console
$ pip install websockets

---> 100%
```

</div>

## WebSockets クライアント { #websockets-client }

    webSocket: WebSocket,
    code: Int,
    reason: String,
  ) {
    Platform.get().log("[WS $name] onClosed $code", Platform.INFO, null)
    val delegate = delegate
    if (delegate != null) {
      this.delegate = null
      delegate.onClosed(webSocket, code, reason)
    } else {
      events.add(Closed(code, reason))
    }
  }

  override fun onFailure(
    webSocket: WebSocket,
    t: Throwable,

# WebSocket 테스트하기 { #testing-websockets }

같은 `TestClient`를 사용하여 WebSocket을 테스트할 수 있습니다.

이를 위해 `with` 문에서 `TestClient`를 사용하여 WebSocket에 연결합니다:

{* ../../docs_src/app_testing/tutorial002_py310.py hl[27:31] *}

/// note

자세한 내용은 Starlette의 [WebSocket 테스트](https://www.starlette.dev/testclient/#testing-websocket-sessions) 문서를 확인하세요.

]

app = FastAPI()


@app.websocket("/function-scope")
async def function_scope(websocket: WebSocket, session: SessionFuncDep) -> Any:
    await websocket.accept()
    await websocket.send_json({"is_open": session.open})


@app.websocket("/request-scope")
async def request_scope(websocket: WebSocket, session: SessionRequestDep) -> Any:
    await websocket.accept()
    await websocket.send_json({"is_open": session.open})

# WebSockets

When defining WebSockets, you normally declare a parameter of type `WebSocket` and with it you can read data from the client and send data to it.

Read more about it in the [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/)

It is provided directly by Starlette, but you can import it from `fastapi`:

```python
from fastapi import WebSocket
```

/// tip

async def get_cookie_or_token(
    websocket: WebSocket,
    session: str | None = Cookie(default=None),
    token: str | None = Query(default=None),
):
    if session is None and token is None:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return session or token


@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    item_id: str,

):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(
            f"Session cookie or query token value is: {cookie_or_token}"
        )
        if q is not None:
            await websocket.send_text(f"Query parameter q is: {q}")