标签: faust

我正在尝试将 Faust 与 Django 集成以将消息发布到 Kafka。这是 Faust 仓库中的示例：https : //github.com/robinhood/faust/tree/master/examples/django

我稍微修改了一下，并创建了视图以通过 Faust 将数据推送到 Kafka。

from django.shortcuts import render

from asgiref.sync import async_to_sync

from accounts.agents import AccountRecord, add_account


async def send_data() -> None:
    print("sending..data")
    print(await add_account.ask(AccountRecord(name="tesst", score=10.9, active=False)))

def index(request):
    async_to_sync(send_data)()
    return render(request, "accounts/index.html")

但是，我现在收到此错误：

RuntimeError at / Task <Task pending name='Task-1' coro=<AsyncToSync.main_wrap() running at /Users/mysuer/.pyenv/versions/3.8.3/envs/faustdjango/lib/python3.8/site-包/asgiref/sync.py:204> cb=[_run_until_complete_cb() at /Users/mysuer/.pyenv/versions/3.8.3/lib/python3.8/asyncio/base_events.py:184]> 将 Future 附加到不同的循环

我正在使用开发服务器运行这个 Django 应用程序。我究竟做错了什么？任何人？:)

我很好奇您应该如何表达您希望将消息快速传递到 Kafka 主题。他们自述文件中的示例似乎没有写入主题：

import faust

class Greeting(faust.Record):
    from_name: str
    to_name: str

app = faust.App('hello-app', broker='kafka://localhost')
topic = app.topic('hello-topic', value_type=Greeting)

@app.agent(topic)
async def hello(greetings):
    async for greeting in greetings:
        print(f'Hello from {greeting.from_name} to {greeting.to_name}')

@app.timer(interval=1.0)
async def example_sender(app):
    await hello.send(
        value=Greeting(from_name='Faust', to_name='you'),
    )

if __name__ == '__main__':
    app.main()

我希望hello.send在上面的代码中向该主题发布一条消息，但似乎没有。

有很多从主题中读取的示例，以及许多使用 cli 推送 ad-hoc 消息的示例。梳理文档后，我没有看到任何明确的代码发布主题示例。我只是疯了，上面的代码应该可以工作吗？

我有一个简单的应用程序，有两个功能，一个用于收听主题，另一个用于 Web 端点。我想创建服务器端事件流 (SSE)，即文本/事件流，以便在客户端我可以使用 EventSource 收听它。

我现在有以下代码，其中每个函数都在执行其特定的工作：

import faust

from faust.web import Response

app = faust.App("app1", broker="kafka://localhost:29092", value_serializer="raw")
test_topic = app.topic("test")


@app.agent(test_topic)
async def test_topic_agent(stream):
    async for value in stream:
        print(f"test_topic_agent RECEIVED -- {value!r}")
        yield value


@app.page("/")
async def index(self, request):
    return self.text("yey")

现在，我想要在索引中，像这样的代码，但使用 faust：

import asyncio
from aiohttp import web
from aiohttp.web import Response
from aiohttp_sse import sse_response
from datetime import datetime


async def hello(request):
    loop = request.app.loop
    async with sse_response(request) as resp:
        while True:
            data = 'Server Time …

我正在尝试使用 pytest 为我的 Faust 应用程序编写单元测试。我已经参考了此处的文档，但它没有提到当我的 Faust 代理将数据发送到接收器时要做什么。

如果没有水槽，我的测试工作正常，但是当我使用水槽时，我收到此错误：

RuntimeError: Task <Task pending name='Task-2' coro=<Agent._execute_actor() running at /Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/site-packages/faust/agents/agent.py:647> cb=[<TaskWakeupMethWrapper object at 0x7fc28967c5b0>()]> got Future <Future pending> attached to a different loop
INFO     faust.agents.agent:logging.py:265 [^-AgentTestWrapper: ml_exporter.processDetections]: Stopping...

我尝试了各种方法：例如修补我的 Faust 应用程序中将数据发送到接收器的装饰器，尝试在没有装饰器的情况下测试我的函数（通过尝试绕过它），修补我的 Faust 应用程序中的接收器参数以有 None 值（因此它不会将我的数据发送到接收器）等。我对这些都没有运气。

这是我的浮士德经纪人：

app = faust.App('ml-exporter', broker=dx_broker, value_serializer='json')

detection_topic = app.topic(dx_topic)
graph_topic = app.topic(gwh_topic)

@app.agent(detection_topic, sink=[graph_topic])
async def processDetections(detections):
    detection_count = 0
    async for detection in detections:
        detection_count += 1
        # r.set("detection_count", detection_count)
        yield detection

这是我当前的测试代码：

import ml_exporter …

我找不到任何答案：Faust和kafka-python有什么区别？
选择其中任何一个有什么优点/缺点吗？
据我了解：

• Kafka是用Java编写的，Kafka-python是一个与“Java流”通信的Python客户端
• Faust是一个纯粹的“Python 流”