websocketsで取得したメッセージをforで表示したい

Question

実行環境

• Python 3.8.2
• websockets 8.1

実現したいこと

websocketsライブラリにおいて、asyncioを隠蔽したいです。

以下コードの for で~~'piyo'~~受信メッセージを表示したいです。
実際にはwebsocketでの受信を逐次表示するような用途で使用したいです。
hogera()より上は、実際にはライブラリ内に入るイメージです。

• 記載変更1 websockets を利用する場合、最初のコードでは全く意味がないと思われたので現時点でのコードを追加しました。

• 記載変更2 最低限守りたいコードが以下です。

for msg in Hoge:
    print(msg)

• 記載変更3 実際には自作のライブラリに入ります。

ライブラリ使用者からasyncioを隠蔽するのが目的です。

現状参考コード

import websockets
import asyncio
import json

class Hoge(object):
    async def stream(self):
        async with websockets.connect(self.uri) as ws:
            while not ws.closed:
                response = await ws.recv()
                self.message = json.loads(response)

    def __init__(self):
        self.message = None

    def __iter__(self):
        return self

    def __next__(self):
        return self.message

''' いずこへ…
        loop = asyncio.get_event_loop()
        loop.create_task(self.stream())
        loop.run_forever()
'''

for msg in Hoge:
    print(msg)

最初に投稿したコード

import asyncio

async def hoge():
    while(True):
        fuga = 'piyo'

def hogera():
    loop = asyncio.get_event_loop()
    loop.create_task(hoge())
    loop.run_forever

msgs = hogera()
for msg in msgs:
    print(msg)  #ここでpiyopiyo言ってほしい

Comments

[toast-uz]

現在の実行結果を記載下さい

[shirasublue]

現在のコードは参考としての掲載で、動作を意図していません。

Traceback (most recent call last):
File "teratail.py", line 29, in <module>
for msg in Hoge:
TypeError: 'type' object is not iterable

[toast-uz]

期待される動作としては、「websocketsに接続しっぱなしにして、非同期に受信するメッセージ群を、for文で逐次表示させたい」ということでしょうか？

[shirasublue]

仰る通りです。

Answer 1

このようなコードで質問者様の意図は実現できるようです。
「非同期に受信するメッセージ群を、for文で処理する」には、for文そのものを非同期にする必要があります。

websocketsライブラリでコネクションを確立したまま持ち回るコードサンプルがあまりなく、苦労しました。
以下のサンプルを見ながら作りました。
websocket_connector.py
なお、json変換はしていません。

import websockets
import asyncio

class Hoge:
    async def connect(self, uri):
        self.ws : websockets.WebSocketCommonProtocol = await websockets.connect(uri)

    def __aiter__(self):
        return self

    async def __anext__(self):
        response = await self.ws.recv()
        return response

async def main():
    hoge = Hoge()
    await hoge.connect('ws://localhost:8000/')
    async for msg in hoge:
        print(msg)

asyncio.run(main())

Comments

[shirasublue]

ありがとうございます。
目的としてはクラス使用者からのasyncioの隠蔽ですが、貴殿のコードでもスマートな書き方が出来そうなので、もし不可能そうであれば貴殿の方法の使用を検討します。

[toast-uz]

同期処理である普通のforループで、非同期で散発的に発生するメッセージを渡すことは、原理的にできません。その時点で到着済のメッセージだけを吐き出すようなforループ（多くの場合は何も吐き出されない）なら作成可能です。利用者からはもう1つ外にwhile Trueの無限ループを作って、sleepを入れてぐるぐるまわす形になります。そういうものをお望みでしょうか？

[toast-uz]

その場合、class全体をイテレーションするような現状の書き方ではなく、到着済メッセージを管理するqueueだけを利用側に見せる実装になると思います。当初に確認していただいた「websocketsに接続しっぱなしにして、非同期に受信するメッセージ群を、for文で逐次表示させたい」からは実装方針が大きく離れると思います。

[shirasublue]

仰るとおり、ホットループが発生するような実装で好ましくないと判断したため、デコレータによる実装へと方針を変えました。ご検討ありがとうございました。

[toast-uz]

asyncioを使う以上は非同期処理が露出してしまうのをどう解決するか悩ましかったのですが、デコレータという解決策は参考になります。こちらこそありがとうごさいます。

Answer 2

質問主題から離れてしまいましたが、asyncio の隠蔽のためデコレータによる flask のような形のコードとしました。
（以下サンプルコードは1ファイルのため asyncio が入っていますが、自作ライブラリに搭載する形で問題なく動作しました。）

import websockets
import asyncio

class EntitySpec(object):
    def __init__(self, ctx):
        self.ctx = ctx
        self.uri = "ws://{}:{}".format(
            ctx.hostname, ctx.port,
        )

    def __call__(self, func):
        async def stream(func):
            async with websockets.connect(self.uri) as ws:
                while not ws.closed:
                    try:
                        response = await ws.recv()
                        func(response)
                    except websockets.exceptions.ConnectionClosedOK:
                        self.loop.stop()
        self.loop = asyncio.get_event_loop()
        self.loop.create_task(stream(func))
        return stream

    def run(self):
        self.loop.run_forever()

class Context(object):
    def __init__(
        self,
        hostname='localhost',
        port=8765,
    ):
        self.hostname = hostname
        self.port = port        
        self.websocket = EntitySpec(self)

api = Context()

@api.websocket
def reciever(msg):
    print(msg)

api.websocket.run()