Alternative/lower-level API

[davidbrochart]

From the documentation, the API consists of a context manager that takes care of opening and closing the connection:

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

What if one wants to open a connection, send/receive data, and close the connection, in separate functions?

[frankie567]

Well, didn't really thought about that 🤔 The key thing is to instantiate a AsyncWebSocketSession class. Problem is that a context manager is needed since we need to make a call to the .stream method of HTTPX, which is also a context manager.

A good trick I learned while developing this library is to use contextlib.AsyncExitStack. It wraps context manager(s) into a single object. So we could probably do something like this:

import contextlib
from httpx_ws import aconnect_ws

# Create the stack
stack = contextlib.AsyncExitStack()
ws = await stack.enter_async_context(aconnect_ws("http://localhost:8000/ws"))

# Work with your session
message = await ws.receive_text()
print(message)

# Close the stack
await stack.aclose()

Not sure if it's the best way to do. I'll try to see if the underlying API could be improved for this.

[davidbrochart]

This seems good enough, thanks a lot!