Add websocket constructor with proxy

I hope I'm not mistaken. With your library we can create a websocket connection with a proxy and, for example, attach it to some synchronous (multi-threaded) or asynchronous function.

Do you have plans to add some high-level constructors for more intuitive use of the websocket client? For example, I want to use something like https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientSession.ws_connect

So far I have poorly understood how to elegantly use a proxied websocket connection that can be used at different runtimes and at the same time used in different forms - a context manager or as a monitor function or in your form that is understandable to me - in an imperative style https://niquests.readthedocs.io/en/latest/user/quickstart.html#synchronous .

more intuitive use of the websocket client?

We kept the ease of use behind Requests. We do not plan to ship such method / function in the near future.

What you seek is easily doable, unless we missed something in your reasoning.

from __future__ import annotations

import typing
from contextlib import asynccontextmanager

import niquests
from urllib3.contrib.webextensions._async import AsyncWebSocketExtensionFromHTTP
import asyncio

@asynccontextmanager
async def ws_connect(url: str, **kwargs) -> typing.AsyncContextManager[AsyncWebSocketExtensionFromHTTP]:
    """Shortcut to WebSocket?"""
    async with niquests.AsyncSession() as s:
        resp = await s.get(url, **kwargs)

        if resp.extension is None:
            raise Exception

        yield resp.extension

        if resp.extension.closed is False:
            await resp.extension.close()

async def run():
    async with ws_connect("wss://echo.websocket.org/", proxies={}, timeout=1) as wss:
        # receive the greeting.
        msg = await wss.next_payload()
        print(msg)

        # Send a TextMessage
        await wss.send_payload("Hello World")

        await wss.ping()  # we can send ping whenever we want,
        # in opposition of ping received, they are handled automatically.
        # Pong are sent automatically.

        # (echo back) Receive a TextMessage
        msg = await wss.next_payload()
        print(msg)  # type = str

        # Send a BinaryMessage
        await wss.send_payload(b"Bytes Msg!")

        # (echo back) Receive a BinaryMessage
        msg = await wss.next_payload()
        print(msg)  # type = bytes

        try:
            await wss.next_payload()
        except niquests.ReadTimeout as e:
            print("Expected, there are no message left to be echoed!", e)

if __name__ == "__main__":
    asyncio.run(run())

Feel free to ping us if this isn't right.

jawah / niquests

Add websocket constructor with proxy #164