Подтвердить что ты не робот

Как я могу реализовать asyncio websockets в классе?

Я хотел бы подключиться к websocket через asyncio и websockets с форматом, как показано ниже. Как я смогу это сделать?

from websockets import connect


class EchoWebsocket:

    def __init__(self):
        self.websocket = self._connect()

    def _connect(self):
        return connect("wss://echo.websocket.org")

    def send(self, message):
        self.websocket.send(message)

    def receive(self):
        return self.websocket.recv()

echo = EchoWebsocket()
echo.send("Hello!")
print(echo.receive())  # "Hello!"

Ответ 1

Как написать асинхронные программы?

Вы должны определить асинхронные функции с помощью async
Вы должны вызвать async funcs с await
Вам нужно цикл событий для запуска вашей асинхронной программы

Все остальные почти такие же, как с обычными программами Python.

import asyncio
from websockets import connect


class EchoWebsocket:
    async def __aenter__(self):
        self._conn = connect("wss://echo.websocket.org")
        self.websocket = await self._conn.__aenter__()        
        return self

    async def __aexit__(self, *args, **kwargs):
        await self._conn.__aexit__(*args, **kwargs)

    async def send(self, message):
        await self.websocket.send(message)

    async def receive(self):
        return await self.websocket.recv()


async def main():
    async with EchoWebsocket() as echo:
        await echo.send("Hello!")
        print(await echo.receive())  # "Hello!"


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Вывод:

Hello!

Как вы видите, код почти такой же, как вы писали.

Единственное отличие состоит в том, что websockets.connect предназначен для работы с асинхронным контекстным менеджером (он использует __aenter__, __aexit__). Необходимо освободить соединение, а также поможет вам выполнять асинхронные операции во время инициализации класса (поскольку у нас нет асинхронной версии __init__).

Я советую вам организовать свой класс таким же образом. Но если вы действительно не хотите использовать контекстный менеджер по какой-либо причине, вы можете использовать новый метод __await__ для инициализации async и некоторой другой функции async для освобождения соединения:

import sys
import asyncio
from websockets import connect


class EchoWebsocket:
    def __await__(self):
        # see: http://stackoverflow.com/a/33420721/1113207
        return self._async_init().__await__()

    async def _async_init(self):
        self._conn = connect("wss://echo.websocket.org")
        self.websocket = await self._conn.__aenter__()
        return self

    async def close(self):
        await self._conn.__aexit__(*sys.exc_info())

    async def send(self, message):
        await self.websocket.send(message)

    async def receive(self):
        return await self.websocket.recv()


async def main():
    echo = await EchoWebsocket()
    try:
        await echo.send("Hello!")
        print(await echo.receive())  # "Hello!"
    finally:
        await echo.close()


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

Многие примеры использования websockets вы можете найти в нем docs.