Shiny for Python：实现一个异步迭代器（几乎就在那里）

Question

最终目标是让应用程序对非阻塞信息流做出反应（在我的特定情况下是 MongoDB ChangeSteam；它也可能是 Kafka 消费者）。

为了可重复性，在下面的示例中，我实现了一个模拟数据流行为的通用异步迭代器AsyncIteratorDummy：

import asyncio
from shiny import reactive, ui, Inputs, Outputs, Session, App, render

class AsyncIteratorDummy:
''' Iterate over an asynchronous source n Iterations.'''

    def __init__(self, n):
        self.current = 0
        self.n = n
    def __aiter__(self):
        return self
    async def __anext__(self):
        await asyncio.sleep(1)
        print(f"get next element {self.current}")
        self.current += 1
        if self.current > self.n:
            raise StopAsyncIteration
        return self.current - 1

async def watch_changes(rval: reactive.Value):
    async for i in AsyncIteratorDummy(5):
        print(f"next element {i}")
        rval.set(i)
           

app_ui = ui.page_fluid(
    "This should update automatically",
    ui.output_text_verbatim("async_text"),
)

def server(input: Inputs, output: Outputs, session: Session):

    triggered_val = reactive.Value(-1)
    
    asyncio.create_task(watch_changes(triggered_val))
    
    @output(id="async_text")
    @render.text()
    async def _():
        return triggered_val.get()
    
    # un/commenting this makes makes the invalidation
    # of `triggered_val` effective or not:
    @reactive.Effect
    def _():
        reactive.invalidate_later(0.1)

app = App(app_ui, server)

该应用程序作品因为存在

    @reactive.Effect
    def _():
        reactive.invalidate_later(0.1)

否则，async_text 变灰（表示已失效）但不更新。

是否可以在没有 reactive.Effect 循环失效的“hack”的情况下实现异步迭代？

我的假设是我必须在watch_changes()（在rval.set(i) 之后）的上下文中“刷新”或“执行”无效变量，使用我无法理解的低级py-shiny 函数。

Answer 1

我想你正在寻找reactive.flush()。

async def watch_changes(rval: reactive.Value):
    async for i in AsyncIteratorDummy(5):
        print(f"next element {i}")
        rval.set(i)
        reactive.flush()