DOCSEARCHbyAlgolia
Submit Docs

httpx-ws

repository·main·Indexed 41 days ago

https://github.com/frankie567/httpx-ws

A library adding WebSocket support to the HTTPX ecosystem. It provides synchronous and asynchronous clients (WebSocketClient, AsyncWebSocketClient) and session managers (WebSocketSession, AsyncWebSocketSession) with built-in helpers for text, binary, and JSON data. Features include automatic lifecycle management for pings/pongs, support for asyncio and Trio via AnyIO, and an ASGIWebSocketTransport for testing WebSockets in ASGI applications like FastAPI or Starlette.

Tokens
8.1K
Snippets
29
Records
35
Agent score
100%

What's inside httpx-ws

  1. Overview of httpx-ws features

    main

    httpx-ws provides WebSocket support for the HTTPX library. Key features include:

    • Sync and async clients: Support for both synchronous and asynchronous workflows.
    • Data helpers: Built-in methods to easily send and receive text, binary, and JSON data.
    • Automatic lifecycle management: Handles automatic ping/pong responses and automatic keepalive pings.
    • Testing support: Includes an HTTPX transport specifically designed to test WebSockets defined in ASGI applications.
    • Async runtime support: Uses AnyIO to provide support for both asyncio and Trio runtimes.

  2. Use WebSocket subprotocols with httpx-ws

    main

    WebSocket subprotocols allow the client and server to agree on a specific communication formalism. When opening a connection with httpx-ws, you can provide a list of supported subprotocols via the subprotocols argument. After the connection is established, you can check which subprotocol the server selected by accessing the subprotocol attribute on the WebSocket object.

    Supported methods:

    • Sync: Use connect_ws within a with block.
    • Async: Use aconnect_ws within an async with block.
    ### Sync Example
```py
from httpx_ws import connect_ws

with connect_ws(
    "http://localhost:8000/ws",
    subprotocols=["my_protocol_1", "my_protocol_2"],
) as ws:
    print(f"Accepted subprotocol {ws.subprotocol}")

    Async Example

    from httpx_ws import aconnect_ws

async with aconnect_ws(
    "http://localhost:8000/ws",
    subprotocols=["my_protocol_1", "my_protocol_2"],
) as ws:
    print(f"Accepted subprotocol {ws.subprotocol}")

  3. Connect to WebSockets synchronously with connect_ws

    main

    Use connect_ws to establish a synchronous WebSocket connection. It returns a WebSocketSession object within a context manager. The session provides helper methods like receive_text() to read messages and send_text() to send messages.

    from httpx_ws import connect_ws

with connect_ws("http://localhost:8000/ws") as ws:
    message = ws.receive_text()
    print(message)
    ws.send_text("Hello!")

  4. Test ASGI apps with ASGIWebSocketTransport

    main

    To test WebSockets defined in an ASGI application (like Starlette or FastAPI) without a running server, use the ASGIWebSocketTransport provided by httpx-ws.

    1. Instantiate ASGIWebSocketTransport passing your ASGI app instance as the argument.
    2. Pass this transport to an httpx.AsyncClient.
    3. Use aconnect_ws to establish WebSocket connections, ensuring you pass the client instance as an argument to aconnect_ws so that WebSocket traffic is routed through the ASGI transport.
    import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport

# Assuming 'app' is your ASGI application instance
async with httpx.AsyncClient(transport=ASGIWebSocketTransport(app)) as client:
    # Standard HTTP requests work normally
    http_response = await client.get("http://server/http")
    assert http_response.status_code == 200

    # WebSocket connections require passing the 'client' to aconnect_ws
    async with aconnect_ws("http://server/ws", client) as ws:
        message = await ws.receive_text()
        assert message == "Hello World!"

  5. Connect to WebSockets asynchronously with aconnect_ws

    main

    Use aconnect_ws for asynchronous WebSocket communication. It returns an AsyncWebSocketSession object within an async with context manager. You must await the message receiving and sending operations.

    from httpx_ws import aconnect_ws

async with aconnect_ws("http://localhost:8000/ws") as ws:
    message = await ws.receive_text()
    print(message)
    await ws.send_text("Hello!")

  6. Develop with httpx-ws

    main

    The development environment is managed using uv for dependency and build management, and just for command shortcuts. Ensure both uv and just are installed on your system.

    Run unit tests

    To execute the full test suite, use:

    just test

    Format the code

    To apply linting and perform type checking, use:

    just lint

  8. Integrate WebSocketClient into a Service SDK using dependency injection

    main

    The class-based APIs (WebSocketClient and AsyncWebSocketClient) are designed to work well with dependency injection. You can pass the client instance into the constructor of your service or SDK class, allowing the service to manage WebSocket connections internally while sharing the underlying HTTPX client lifecycle.

    import httpx
from httpx_ws import AsyncWebSocketClient

# Your service SDK
class MyServiceSDK:
    def __init__(self, websocket_client: AsyncWebSocketClient) -> None:
        self._websocket_client = websocket_client

    async def hello(self) -> str:
        async with self._websocket_client.connect("ws://localhost:8000/ws") as ws:
            await ws.send_text("Hello!")
            response = await ws.receive_text()
            return response

# Usage example
async def main() -> None:
    async with httpx.AsyncClient() as client:
        ws_client = AsyncWebSocketClient(client)

        service = MyServiceSDK(ws_client)
        response = await service.hello()
        print(f"Received: {response}")

  9. Use an existing httpx Client with connect_ws or aconnect_ws

    main

    To customize connection parameters or leverage connection pooling, you can pass an existing httpx.Client (for sync) or httpx.AsyncClient (for async) to the connection functions.

    import httpx
from httpx_ws import connect_ws

with httpx.Client() as client:
    with connect_ws("http://localhost:8000/ws", client) as ws:
        message = ws.receive_text()
        print(message)
        ws.send_text("Hello!")
    import httpx
from httpx_ws import aconnect_ws

async with httpx.AsyncClient() as client:
    async with aconnect_ws("http://localhost:8000/ws", client) as ws:
        message = await ws.receive_text()
        print(message)
        await ws.send_text("Hello!")

  10. Use the synchronous WebSocketClient

    main

    For synchronous workflows, use the WebSocketClient class. It wraps an existing httpx.Client instance. You can establish a connection using the .connect(url) method within a context manager, which provides a WebSocket object for sending and receiving messages.

    import httpx
from httpx_ws import WebSocketClient

with httpx.Client() as client:
    ws_client = WebSocketClient(client)
    with ws_client.connect("http://localhost:8000/ws") as ws:
        message = ws.receive_text()
        print(message)
        ws.send_text("Hello!")

  11. Use the asynchronous AsyncWebSocketClient

    main

    For asynchronous workflows, use the AsyncWebSocketClient class. It wraps an existing httpx.AsyncClient instance. Use async with to manage both the client and the connection established via .connect(url). The resulting WebSocket object supports asynchronous methods like await ws.receive_text() and await ws.send_text().

    import httpx
from httpx_ws import AsyncWebSocketClient

async with httpx.AsyncClient() as client:
    ws_client = AsyncWebSocketClient(client)
    async with ws_client.connect("http://localhost:8000/ws") as ws:
        message = await ws.receive_text()
        print(message)
        await ws.send_text("Hello!")

  12. Handle WebSocket connection errors and denials

    main

    When testing WebSockets with httpx-ws, you may encounter specific exceptions related to the connection lifecycle:

    • WebSocketUpgradeError: Raised if the ASGI application sends a websocket.http.response.start message instead of accepting the connection. This error wraps the httpx.Response containing the status code, headers, and body of the denial.
    • WebSocketDisconnect: Raised when the ASGI application sends a websocket.close message. It contains the code and reason provided by the application.
    • UnhandledASGIMessageType: Raised if the stream receives an ASGI message type that is not websocket.send or websocket.close during a read operation.
    • UnhandledWebSocketEvent: Raised if the underlying wsproto connection produces an event that the transport does not recognize.