httpx-ws 0.1.0

WebSockets support for HTTPX

httpx-ws

WebSockets support for HTTPX

Installation

Not released on PyPI yet 😅

pip install httpx-ws

Quickstart

httpx-ws provides connect_ws and aconnect_ws to help connecting and communication with WebSockets. The resulting WebSocketSession/AsyncWebSocketSession object provides helpers to send and receive messages in the WebSocket.

Sync

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!")

Async

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!")

You can also pass an httpx.Client/httpx.AsyncClient if you want to customize parameters or benefit from connection pooling:

Sync

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!")

Async

import httpx
from httpx_ws import aconnect_ws

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!")

Testing ASGI apps

You can use httpx_ws to test WebSockets defined in an ASGI app, just like you do with HTTPX for HTTP endpoints.

For this, we've implemented a custom transport for HTTPX, ASGIWebSocketTransport. You need to instantiate a class of this transport and set it as parameter on your HTTPX client.

Let's say you have this Starlette app:

from starlette.applications import Starlette
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute


async def http_hello(request):
    return HTMLResponse("Hello World!")

async def ws_hello(websocket):
    await websocket.accept()
    await websocket.send_text("Hello World!")
    await websocket.close()


app = Starlette(
    routes=[
        Route("/http", http_hello),
        WebSocketRoute("/ws", ws_hello),
    ],
)

You can call it directly like this:

import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport

async with httpx.AsyncClient(transport=ASGIWebSocketTransport(app)) as client:
    http_response = await client.get("http://server/http")
    assert http_response.status_code == 200

    async with aconnect_ws("http://server/ws", client) as ws:
        message = await ws.receive_text()
        assert message == "Hello World!"

Notice that, in this case, you must pass the client instance to aconnect_ws. HTTP requests are handled normally.

Development

Setup environment

We use Hatch to manage the development environment and production build. Ensure it's installed on your system.

Run unit tests

You can run all the tests with:

hatch run test

Format the code

Execute the following command to apply isort and black formatting:

hatch run lint

License

This project is licensed under the terms of the MIT license.