WebSockets support for HTTPX
⚠️ This is a very young project. Expect bugs 🐛
pip install httpx-wshttpx-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!")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.
We use Hatch to manage the development environment and production build. Ensure it's installed on your system.
You can run all the tests with:
hatch run testExecute the following command to apply isort and black formatting:
hatch run lintThis project is licensed under the terms of the MIT license.
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent