WebSockets support for HTTPX
Project description
httpx-ws
WebSockets support for HTTPX
Installation
⚠️ This is a very young project. Expect bugs 🐛
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.
Contributors ✨
Thanks goes to these wonderful people (emoji key):
François Voron 🚧 💻 |
Kousik Mitra 💻 |
This project follows the all-contributors specification. Contributions of any kind welcome!
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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.