WebSockets support. #304
Comments
|
So my assumption here would be "no", but that it's possible we'll want to expose just enough API in order to allow a websocket implementation to use httpx for handshaking the connection, and sharing the connection pool. So, to the extent that we might allow it to be possible, I'd expect that to be in the form of a third party package, that has I'm not sure exactly what API that'd imply that we'd need to support, but it's feasible that we might end up wanting to expose some low-level connection API specifically for supporting this kind of use case. The most sensible tack onto thrashing out what we'd need there would be a proof-of-concept implementation that'd help us concretely indentify how much API we'd need to make that possible. Does all that make sense, or am I off on the wrong track here? |
|
What you are saying does make sense, I haven't quite made up my own mind on this. Pros of a third-party package handling the websockets:
Cons:
|
|
Interesting points, yeah. I guess I'd be open to reassessing this as we go. We'd need to identify what the API would look like, both at the top user-facing level, and at whatever new lower-level cutoffs we'd need to be introducing. |
|
I'm in favor of adding websockets definitely, would be a good feature to work towards in a v2? Still so much to do to get a v1 released. Thanks for bringing this up @jlaine! I've not had to deal with websockets so I didn't even think they'd be at home in a library like HTTPX. :) |
Probably not a bad call, yeah. |
I'll see if I can find time to put together a minimal PR supporting HTTP/1.1 and HTTP/2 to scope out what this entails. Obviously if there are more pressing needs for v1 there is zero pressure to merge it. @aaugustin just pinging you so you're in the loop : this is not a hostile move against |
|
I'm currently working on refactoring websockets to provide a sans-I/O layer. If it's ready within a reasonable timeframe, perhaps you could use it. It will handle correctly a few things that wsproto doesn't, if the comments I'm seeing in wsproto's source code are correct. One of my issues with sans-I/O is the prohibition async / await, making it impossible to provide a high-level API. For this reason, I'm not sure that sans-I/O as currently specified is the best approach. This makes me hesitate on a bunch of API design questions... Building an API that httpx will consume (either strict sans-I/O or something else) would be a great step forwards. Perhaps we can cooperate on this? Also I would really like I'm quite happy with my implementation of a subset of HTTP/1.1 — I'm not kidding, even though it's home-grown, I honestly think it's all right. It may be possible to achieve HTTP/2 in the same way but it will be partial e.g. it will be impossible to multiplex websockets and HTTP on the same connection. Given what I've seen of aioquic, I don't think it's reasonable to do HTTP/3 that way. So I'm interested in working with third-party packages that handle HTTP/2 and HTTP/3 to figure out what that entails. |
|
As for me, httpx is not only the next generation HTTP client for requests but also aiohttp. So expecting a new choice to take the place of aiohttp's websocket. https://github.com/ftobia/aiohttp-websockets-example/blob/master/client.py Best wishes to the author, this lib is a great help for me. PS: aiohttp is good enough, except some frequently changing api protocol (like variable names), which raised a backward incompatibility error by a new version. |
|
FYI I started implementing a Sans-I/O layer for websockets: python-websockets/websockets#466 It may not be obvious from the first commits because I have to untangle a lot of stuff from asyncio before I can get anything done :-) |
|
@aaugustin That's awesome news! :) Thanks for updating this thread. It'd be interesting to see how an upgrade from (Also, congrats on the Tidelift sponsorship!) |
|
So, here's how I think the bridge could work. Since we're talking about httpx, I'm focusing on the client side here, but the same logic would apply on the server side. In addition to the regular APIs that receive and send bytes, websockets should provide an API to receive and send already parsed HTTP headers. This API will support bolting websockets on top of any HTTP/1.1 implementation. In addition to: from websockets import ClientConnection
ws = ClientConnection()
bytes_to_send = ws.connect()
...
ws.receive_data(bytes_received)I need to provide something like: from websockets import ClientConnection
ws = ClientConnection()
request = ws.build_handshake_request()
# here request is (path, headers) tuple
bytes_to_send = serialize(request) # <-- you can do this with any library you want
...
response = parse(bytes_received) # <-- you can do this with any library you want
# here response is a (status_code, reason_phrase, headers) tuple
ws.receive_handshake_response()The latter happens under the hood anyway so it's clearly doable. It's "just" a matter of naming things :-) — which I care deeply about. WebSocket over HTTP/2 requires a different handshake, so websockets will need another APIs to support it. I have no idea about WebSocket over HTTP/3. |
|
WebSockets over HTTP/3 haven't been officially specified, but it is extremely likely it will work like for HTTP/2 (RFC 8441), namely using a |
|
I'd like us to treat this as out-of-scope at this point in time. Yes I think we'll want to provide enough API to make this do-able at some point in the not too distant future, but being able to do that while still aiming for a sqaured away API stable 1.0 release isn't something we're able to do just yet. |
|
I'd recommend to keep this open and add a "deferred" label. |
|
It might also be worth tweaking the AsyncClient documentation; currently, it says:
and indeed httpx pops up as one of the first results on Google for |
|
So, I've been thinking a bit about how we might support websockets and other upgrade protocols from This is more of a rough idea, than a formal plan at the moment, but essentially what we'll want is for the
Which will return an object implementing our (currently internal only) SocketStream API Once we're happy that we've got the low-level # Using a `client` instance here, but we might also support a plain `httpx.upgrade(...)` function.
# Both sync + async variants can be provided here.
with client.upgrade(url, "websocket") as connection:
# Expose the fundamental response info...
connection.url, connection.headers, ...
# Also expose the SocketStream primitives...
connection.read(...), connection.write(), ...With HTTP/2, the socket stream would actually wrap up an Protocol libraries wouldn't typically expose this interface themselves, but rather would expose whatever API is appropriate for their own cases, using Nice stuff this would give us...
We don't necessarily want to rush trying to get this done, but @aaugustin's work on python-websockets/websockets#466 has gotten me back to thinking about it, and it feels like quite an exciting prospect. Does this seem like a reasonable way forward here?... /cc @pgjones @florimondmanca @jlaine @aaugustin @sethmlarson |
|
The Sans I/O layer in websockets is (AFAIK) feature complete with full test coverage. However, websockets doesn't uses it yet and I haven't written the documentation yet. If you have a preference between:
I can priorize my efforts accordingly. Also, I don't expect you to use the HTTP/1.1 implementation in websockets, only the handshake implementation (which contains the extensions negotiation logic, and you don't want to rewrite that). You will want to work with request / response abstractions serialized / parsed by httpx rather than websockets. At some point, making this possible was a design goal. I don't swear it is possible right now :-) We can investigate together the cleanest way to achieve this. |
|
In my view supporting sending requests over the same connection as the WebSocket for HTTP/2 is a key feature. As an isolated connection may as well start with the HTTP/1-Handshake and avoid the HTTP/2 extra complexity. I've not followed the httpcore/httpx structure so I can't comment on the details, sorry. I'll also make the case for wsproto, which supports WebSockets, HTTP/1-Handshakes, and HTTP/2-Handshakes. It also works very happily with h11, and h2. It is also now typed and I would say stable - indeed we've been discussing a 1.0 release. |
|
wsproto is perfectly fine as well :-) |
tomchristie
changed the title
|
Thanks folks - not going to act on any of this just yet since it's clearly a post 1.0 issue. In the first pass of this we'd simply be exposing a completely agnostic connection upgrade API, which would allow third party packages to build whatever websocket implementations they want, piggybacking on top of We could potentially also consider built-in websocket support at that point, but let's talk about that a little way down the road. |
florimondmanca
added
discussion
and removed
question
|
Still relevant |
|
Hello! I was looking at using the Just crossposting my discussion since I haven't gotten a response yet and I suspect I am simply approaching this wrong: encode/httpcore#572. |
|
I wonder what's the state of the art this year. I was about to write a small "api server/balancer" with But... it turns out I would need to support a single, pesky websocket endpoint, it too would be basically "connected" to one of the backends. I wonder how I could hack this together. 🤔 |
|
This is about to be merged in httpcore: encode/httpcore#581 |
|
Okay, so since the latest release of This means we're able to read/write directly to upgraded network streams. We can combine this with the import httpx
import wsproto
import os
import base64
url = "http://127.0.0.1:8765/"
headers = {
b"Connection": b"Upgrade",
b"Upgrade": b"WebSocket",
b"Sec-WebSocket-Key": base64.b64encode(os.urandom(16)),
b"Sec-WebSocket-Version": b"13"
}
with httpx.stream("GET", url, headers=headers) as response:
if response.status_code != 101:
raise Exception("Failed to upgrade to websockets", response)
# Get the raw network stream.
network_steam = response.extensions["network_stream"]
# Write a WebSocket text frame to the stream.
ws_connection = wsproto.Connection(wsproto.ConnectionType.CLIENT)
message = wsproto.events.TextMessage("hello, world!")
outgoing_data = ws_connection.send(message)
network_steam.write(outgoing_data)
# Wait for a response.
incoming_data = network_steam.read(max_bytes=4096)
ws_connection.receive_data(incoming_data)
for event in ws_connection.events():
if isinstance(event, wsproto.events.TextMessage):
print("Got data:", event.data)
# Write a WebSocket close to the stream.
message = wsproto.events.CloseConnection(code=1000)
outgoing_data = ws_connection.send(message)
network_steam.write(outgoing_data)(I tested this client against the websockets example server given here... https://websockets.readthedocs.io/en/stable/) This gives us enough that someone could now write an I assume the API would mirror the import httpx
from httpx_websockets import connect_websockets
with httpx.Client() as client:
with connect_websockets(client, "ws://localhost/sse") as websockets:
outgoing = "hello, world"
websockets.send(outgoing)
incoming = websockets.receive()
print(incoming)With both sync and async variants. If anyone's keen on taking a go at that I'd gladly collaborate on it as needed. |
|
Aside: WebSockets over HTTP/2 would require encode/httpcore#592. |
|
FWIW websockets provides a Sans-I/O layer too. It used to be more feature-complete; wsproto caught up in the recent years; I don't know how the two libraries compare these days. Based on the docs, I believe websockets' handling of failure scenarios is more robust i.e. it tells you when things have gone wrong and you should just close the TCP connection (even if you're the client). |
|
One thing that websockets is missing — and that I had in mind when I wrote the Sans-I/O layer — is the ability to handshake over HTTP/2. I believe you have that :-) I'd be interested in figuring out the graft. |
Neato. Step zero would be to ping up a little example similar to the
I'd suppose the benefits of using We could feasibly add websockets-over-http/2 as well, yes, although it'd need a bit more work, see above. |
|
I started to work on this, following what you recommended, @tomchristie: https://github.com/frankie567/httpx-ws The main pain point for me was to actually implement an It's still very experimental, but it's a start 😊 |
|
Here's one way to do it with websockets. A pretty big difference with the example with wsproto above — I'm actually executing the opening handshake logic, meaning that extensions, subprotocols, etc. are negotiated. In practice, this means that compression works. Also, lots of small things could be cleaner: for example, httpx insists on having a import httpx
# ClientConnection is renamed to ClientProtocol in websockets 11.0.
# Sorry, I grew unhappy with my inital attempt at naming things!
# I'm using the future-proof name here.
from websockets.client import ClientConnection as ClientProtocol
from websockets.connection import State
from websockets.datastructures import Headers
from websockets.frames import Opcode
from websockets.http11 import Response
from websockets.uri import parse_uri
def connect_to_websocket(url):
# Force the connection state to OPEN instead of CONNNECTING because
# we're handling the opening handshake outside of websockets.
protocol = ClientProtocol(
parse_uri(url.replace("http://", "ws://")),
state=State.OPEN,
)
# Start WebSocket opening handshake.
request = protocol.connect()
with httpx.stream("GET", url, headers=request.headers) as response:
# Get the raw network stream.
network_steam = response.extensions["network_stream"]
# Convert httpx response to websockets response.
response = Response(
response.status_code,
response.reason_phrase,
Headers(response.headers),
)
# Complete WebSocket opening handshake.
protocol.process_response(response)
# Write a WebSocket text frame to the stream.
protocol.send_text("hello, world!".encode())
for outgoing_data in protocol.data_to_send():
network_steam.write(outgoing_data)
# Wait for a response.
incoming_data = network_steam.read(max_bytes=4096)
protocol.receive_data(incoming_data)
for frame in protocol.events_received():
if frame.opcode is Opcode.TEXT:
print("Got data:", frame.data.decode())
# Write a WebSocket close to the stream.
protocol.send_close()
for outgoing_data in protocol.data_to_send():
network_steam.write(outgoing_data)
if __name__ == "__main__":
connect_to_websocket("http://127.0.0.1:8765/") |
|
Re. asyncio / trio / threads, actually websockets isn't all that well covered:
|
Would someone like to raise an issue so we can resolve that? There's a very minor change needed in We do already handle the correct default port mapping for ws and wss, and @aaugustin - Thanks for your example above! I've reworked that to provide a basic API example... import contextlib
import httpx
from websockets.client import ClientConnection as ClientProtocol
from websockets.connection import State
from websockets.datastructures import Headers
from websockets.frames import Opcode
from websockets.http11 import Response
from websockets.uri import parse_uri
class ConnectionClosed(Exception):
pass
class WebsocketConnection:
def __init__(self, protocol, network_steam):
self._protocol = protocol
self._network_stream = network_steam
self._events = []
async def send(self, data):
self._protocol.send_text("hello, world!".encode())
for outgoing_data in self._protocol.data_to_send():
await self._network_stream.write(outgoing_data)
async def recv(self):
while True:
if not self._events:
incoming_data = await self._network_stream.read(max_bytes=4096)
self._protocol.receive_data(incoming_data)
self._events = self._protocol.events_received()
for event in self._events:
if event.opcode is Opcode.TEXT:
return event.data.decode()
elif event.opcode is Opcode.CLOSE:
raise ConnectionClosed()
@contextlib.asynccontextmanager
async def connect(url):
protocol = ClientProtocol(
parse_uri(url.replace("http://", "ws://")),
state=State.OPEN,
)
# Start WebSocket opening handshake.
request = protocol.connect()
async with httpx.AsyncClient() as client:
async with client.stream("GET", url, headers=request.headers) as response:
# Get the raw network stream.
network_steam = response.extensions["network_stream"]
# Convert httpx response to websockets response.
response = Response(
response.status_code,
response.reason_phrase,
Headers(response.headers),
)
# Complete WebSocket opening handshake.
protocol.process_response(response)
yield WebsocketConnection(protocol, network_steam)You can then use that with import asyncio
async def hello(uri):
async with connect(uri) as websocket:
await websocket.send("Hello world!")
print(await websocket.recv())
asyncio.run(hello("http://localhost:8765"))Or with import trio
async def hello(uri):
async with connect(uri) as websocket:
await websocket.send("Hello world!")
print(await websocket.recv())
trio.run(hello, "http://localhost:8765")To implement the equivalent threaded version, drop the
That's looking pretty neat. It occurs to me that a neater API would be to make the with connect_ws(client, "http://localhost:8000/ws") as ws:
message = ws.receive_text()
print(message)
ws.send_text("Hello!")and if you want shared connection pooling, then... 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!") |
|
Is there a plan to get this built into AsyncClient? I'm struggling a bit to write end-to-end tests where I want to listen to websockets in the background while i post data to fastapi. I will try to work with your examples, but this really is a lot of code that would be nice to have as a method on AsyncClient like Starlette has on it's TestClient |
|
@wholmen You can try the library I've just created: https://github.com/frankie567/httpx-ws In particular, here is how you could set up a test client to test a FastAPI app: https://frankie567.github.io/httpx-ws/usage/asgi/ |
|
So I suppose we should add this and They're such important use cases that I'd probably suggest something like prominent "Websockets" and "Server Sent Events" headings above the existing "Plugins"? |
|
I needed a behavior where I could connect to the websocket and listen to messages from the server indefinitely, until I needed to disconnect, while making sure that every time a message came the program would perform the action I wanted. In this implementation, it didn't work so the |
|
@ll2pakll Thanks for the prompt. I can see the error there now, although the Chat-CPT version isn't quite right either. (Might block reading from the network while there's pending events that could be returned.) Here's an updated |
|
Not websocket support so ? |
Currently
httpxis squarely focused on HTTP's traditional request / response paradigm, and there are well-established packages for WebSocket support such as websockets. In an HTTP/1.1-only world, this split of responsabilities makes perfect sense as HTTP requests / WebSockets work independently.However, with HTTP/2 already widely deployed and HTTP/3 standardisation well under way I'm not sure the model holds up:
websocketsare usually tied to HTTP/1.1 only, whereashttpxhas support for HTTP/2 (and hopefully soon HTTP/3)Using the sans-IO
wsprotocombined withhttpx's connection management we could provide WebSocket support spanning HTTP/1.1, HTTP/2 and HTTP/3. What are your thoughts on this?One caveat: providing WebSocket support would only make sense using the
AsyncClientinterface.The text was updated successfully, but these errors were encountered: