pythonpython-3.xapache-kafkaaiohttpfaust
How to connect kafka topic with web endpoint using Faust Python package?
I have a simple app, with two functions, one for listening to topic and other for web endpoint. I want to create server side event streaming (SSE) i.e text/event-stream, so that on client end I could listen to it using EventSource.
I have the following code for now, where each function is doing its particular job:
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")
Now, I want in the index, something like this code, but using 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 : {}'.format(datetime.now())
print(data)
await resp.send(data)
await asyncio.sleep(1, loop=loop)
return resp
async def index(request):
d = """
<html>
<body>
<script>
var evtSource = new EventSource("/hello");
evtSource.onmessage = function(e) {
document.getElementById('response').innerText = e.data
}
</script>
<h1>Response from server:</h1>
<div id="response"></div>
</body>
</html>
"""
return Response(text=d, content_type='text/html')
app = web.Application()
app.router.add_route('GET', '/hello', hello)
app.router.add_route('GET', '/', index)
web.run_app(app, host='127.0.0.1', port=8080)
I have tried this:
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("/", name="t1")
@app.agent(test_topic, name="t")
async def index(self, request):
return self.text("yey")
But it gives me the following error:
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 299, in find_app
val = symbol_by_name(app, imp=imp)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 262, in symbol_by_name
module = imp( # type: ignore
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 376, in import_from_cwd
return imp(module, package=package)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
File "<frozen importlib._bootstrap>", line 991, in _find_and_load
File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 783, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/Users/maverick/company/demo1/baiohttp-demo/app1.py", line 18, in <module>
async def index(self, request):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/app/base.py", line 1231, in _decorator
view = view_base.from_handler(cast(ViewHandlerFun, fun))
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/web/views.py", line 50, in from_handler
return type(fun.__name__, (cls,), {
AttributeError: 'Agent' object has no attribute '__name__'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/faust_demo/bin/faust", line 8, in <module>
sys.exit(cli())
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/click/core.py", line 829, in __call__
return self.main(*args, **kwargs)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/click/core.py", line 781, in main
with self.make_context(prog_name, args, **extra) as ctx:
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 407, in make_context
self._maybe_import_app()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 372, in _maybe_import_app
find_app(appstr)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/cli/base.py", line 303, in find_app
val = imp(app)
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/utils/imports.py", line 376, in import_from_cwd
return imp(module, package=package)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1014, in _gcd_import
File "<frozen importlib._bootstrap>", line 991, in _find_and_load
File "<frozen importlib._bootstrap>", line 975, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 671, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 783, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "/Users/maverick/company/demo1/baiohttp-demo/app1.py", line 18, in <module>
async def index(self, request):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/app/base.py", line 1231, in _decorator
view = view_base.from_handler(cast(ViewHandlerFun, fun))
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/web/views.py", line 50, in from_handler
return type(fun.__name__, (cls,), {
AttributeError: 'Agent' object has no attribute '__name__'
I event tried this:
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.agent(test_topic, name="t")
@app.page("/", name="t1")
async def index(self, request):
return self.text("yey")
But I get following error:
[2020-03-28 10:32:50,676] [29976] [INFO] [^--Producer]: Creating topic 'app1-__assignor-__leader'
[2020-03-28 10:32:50,695] [29976] [INFO] [^--ReplyConsumer]: Starting...
[2020-03-28 10:32:50,695] [29976] [INFO] [^--AgentManager]: Starting...
[2020-03-28 10:32:50,695] [29976] [INFO] [^---Agent: app1.index]: Starting...
[2020-03-28 10:32:50,696] [29976] [ERROR] [^Worker]: Error: TypeError("__init__() missing 1 required positional argument: 'web'")
Traceback (most recent call last):
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/worker.py", line 273, in execute_from_commandline
self.loop.run_until_complete(self._starting_fut)
File "/Users/maverick/.pyenv/versions/3.8.1/lib/python3.8/asyncio/base_events.py", line 612, in run_until_complete
return future.result()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 767, in _actually_start
await child.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 767, in _actually_start
await child.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 760, in _actually_start
await self.on_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/manager.py", line 58, in on_start
await agent.maybe_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 795, in maybe_start
await self.start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 736, in start
await self._default_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 743, in _default_start
await self._actually_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/mode/services.py", line 760, in _actually_start
await self.on_start()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 282, in on_start
await self._on_start_supervisor()
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 312, in _on_start_supervisor
res = await self._start_one(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 251, in _start_one
return await self._start_task(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 617, in _start_task
actor = self(
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 525, in __call__
return self.actor_from_stream(stream,
File "/Users/maverick/.pyenv/versions/3.8.1/envs/faust_demo/lib/python3.8/site-packages/faust/agents/agent.py", line 552, in actor_from_stream
res = self.fun(actual_stream)
TypeError: __init__() missing 1 required positional argument: 'web'
[2020-03-28 10:32:50,703] [29976] [INFO] [^Worker]: Stopping...
[2020-03-28 10:32:50,703] [29976] [INFO] [^-App]: Stopping...
[2020-03-28 10:32:50,703] [29976] [INFO] [^-App]: Flush producer buffer...
[2020-03-28 10:32:50,703] [29976] [INFO] [^--TableManager]: Stopping...
Could there be a way for this? Thanks a lot in advance!
Solution
The Faust worker will also expose a web server on every instance, that by default runs on port 6066.
The server will use the aiohttp HTTP server library and you can take advantage of this thing and create a server-side event streaming (SSE) like in your example code.
You can create an agent that will read from Kafka topic test and will update a variable last_message_from_topic with the last message from the topic, this variable will be visible also from your web pages.
In the index page (@app.page('/')) the EventSource interface is used to receive server-sent events. It connects to the server over HTTP and receives events in text/event-stream format from the page /hello without closing the connection.
The web page /hello at every second is sending a message text with the last message from the Kafka topic test and with the current time from the server.
here is my file my_worker.py code:
import asyncio
from datetime import datetime
import faust
from aiohttp.web import Response
from aiohttp_sse import sse_response
app = faust.App(
"app1",
broker='kafka://localhost:9092',
value_serializer='json',
)
test_topic = app.topic("test")
last_message_from_topic = ['No messages yet']
@app.agent(test_topic)
async def greet(greetings):
async for greeting in greetings:
last_message_from_topic[0] = greeting
@app.page('/hello')
async def hello(self, request):
loop = request.app.loop
async with sse_response(request) as resp:
while True:
data = f'last message from topic_test: {last_message_from_topic[0]} | '
data += f'Server Time : {datetime.now()}'
print(data)
await resp.send(data)
await asyncio.sleep(1, loop=loop)
return resp
@app.page('/')
async def index(self, request):
d = """
<html>
<body>
<script>
var evtSource = new EventSource("/hello");
evtSource.onmessage = function(e) {
document.getElementById('response').innerText = e.data
}
</script>
<h1>Response from server:</h1>
<div id="response"></div>
</body>
</html>
"""
return Response(text=d, content_type='text/html')
now you have to start the Faust worker with the following command:
faust -A my_worker worker -l info
on your web browser you can access http://localhost:6066/:
here is the code to send messages to Kafka on the topic test (from another python file):
import time
import json
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],value_serializer=lambda x: json.dumps(x).encode('utf-8'))
for i in range(220):
time.sleep(1)
producer.send('test', value=f'Some message from kafka id {i}')
- matplotlib contourf: get Z value under cursor
- Detect if an OCR text image is upside down
- How to sort a list of lists of lists by the total number of items?
- What is the recommended way for retrieving row numbers (index) for polars?
- How to upload an image file to Github using PyGithub?
- How to store and load a Python dictionary with HDF5
- Prevent Selenium from taking the focus to the opened window
- Subclass super().__init__(*args, **kwargs) not working. Says object.__init__() takes exactly one argument (the instance to initialize) when it doesn't
- Overriding python threading.Thread.run()
- When searching for strings in a dataframe via lambda function, how do I resolve the IndexError if no results are found in any column or row?
- ValuerError: Found input variables with inconsistent numbers of samples
- sqlite3 or CSV files
- isPrime Function for Python Language
- Sending data and files as multipart request using aiohttp
- Official repository of Unicode character names
- How to create child window and communicate with parent in TkInter
- with pytest.raises(Exception) not working with flask app.post
- ModuleNotFoundError with pyinstaller but fine in Pycharm
- Storing and retreiving data with Milvus and Langchain
- Python Barplot to represent Ranges
- Docker image python-alpine fails when using lib
- create azure functionapp from docker image via cli
- Routing submodule functions using FastAPI's APIRouter
- clone element with beautifulsoup
- Sharing variables between two seperatly running python scripts
- How do I filter a table by clicking on a bar chart segment?
- pyarrow: find diff for chunkedarray
- Skip statements in case of last item in iterator
- csv file to host.yaml for Nornir
- When I'm plotting a colorbar in python using matplotlib I'm getting an error