feat: minimal working mvp

Author: woile

README.md

@@ -1,2 +1,37 @@
 # dam
+
 Python Stream Processing toolkit
+
+:warning: WARNING :warning:
+
+_This repo is highly experimental and it's not recommended for production use._
+
+_The interface may change at any time during initial development._
+
+```python
+import asyncio
+from dam import App, Topic
+
+loop = asyncio.get_event_loop()
+
+async def hello_world(stream):
+    async for value in stream:
+        print(value.value)
+
+
+topics = [
+    Topic("test-topic-worker", hello_world)
+]
+
+streamapp = App(topics=topics)
+
+while True:
+    # This should be any other worker
+    loop.run_until_complete(streamapp.run())
+```
+
+## Usage
+
+1. Clone repository
+2. Run `poetry install`
+3. Write apps inside `example` folder (for now)

dam/__init__.py

@@ -1 +1,7 @@
+from dam.application import App
+from dam.topic import Topic
+
+__all__ = ["App", "Topic"]
+
+
 __version__ = "0.1.0"

dam/application.py

@@ -0,0 +1,18 @@
+import asyncio
+from typing import List, Optional
+
+
+class App:
+    def __init__(
+        self,
+        topics: Optional[List] = None,
+        loop: Optional[asyncio.AbstractEventLoop] = None,
+    ):
+        if loop is None:
+            loop = asyncio.get_event_loop()
+        self.loop = loop
+        self.topics = topics
+
+    async def run(self):
+        for topic in self.topics:
+            await topic.run()

dam/stream.py

@@ -0,0 +1,10 @@
+class Stream:
+    def __init__(self, stream):
+        self.stream = stream
+
+    async def __aiter__(self):
+        async for item in self.stream:
+            yield item
+
+    async def __anext__(self):
+        raise StopAsyncIteration

dam/topic.py

@@ -0,0 +1,65 @@
+import asyncio
+from typing import Callable
+
+from aiokafka import AIOKafkaConsumer
+
+from dam.stream import Stream
+
+
+class Topic:
+    """Kafka topic"""
+
+    def __init__(self, name: str, func: Callable):
+        self.name = name
+        self.func = func
+
+    def create_consumer(
+        self, name: str, loop: asyncio.AbstractEventLoop = None
+    ) -> AIOKafkaConsumer:
+        if loop is None:
+            loop = asyncio.get_event_loop()
+
+        return AIOKafkaConsumer(
+            name,
+            loop=loop,
+            bootstrap_servers="0.0.0.0:29092",
+            group_id="test-consumer-group",
+            auto_offset_reset="earliest",
+        )
+
+    async def run(self):
+        consumer = self.create_consumer(self.name)
+        await consumer.start()
+
+        try:
+            # Consume messages
+            await self.func(Stream(consumer))
+        finally:
+            # Will leave consumer group; perform autocommit if enabled.
+            await consumer.stop()
+
+
+# async def consume_from_kafka(loop):
+#     logger.debug("Task Consuming from kafka initiated...")
+#     consumer = AIOKafkaConsumer(
+#         TOPIC,
+#         loop=loop,
+#         bootstrap_servers=BOOTSTRAP_SERVERS,
+#         group_id=GROUP_ID,
+#         auto_offset_reset=AUTO_OFFSET_RESET,
+#     )
+
+#     # Get cluster layout and join group `my-group`await consumer.start()
+#     await consumer.start()
+
+#     try:
+#         # Consume messages
+#         async for msg in consumer:
+#             print(f"Task Consuming from kafka")
+#             print(
+#                 f"Topic: {msg.topic}, Partition: {msg.partition}, Offset: {msg.offset},"
+#                 f" Key: {msg.key}, Value: {msg.value}, Timestamp: {msg.timestamp}\n"
+#             )
+#     finally:
+#         # Will leave consumer group; perform autocommit if enabled.
+#         await consumer.stop()

example/simple_consumer.py

@@ -0,0 +1,26 @@
+import asyncio
+from dam import App, Topic
+
+loop = asyncio.get_event_loop()
+
+
+async def hello_world(stream):
+    async for value in stream:
+        print(value.value)
+
+
+async def hello_world2(stream):
+    async for value in stream:
+        print("other topic", value.value)
+
+
+topics = [
+    Topic("test-topic-worker", hello_world),
+    Topic("fifo", hello_world2)
+]
+
+streamapp = App(topics=topics)
+
+while True:
+    # This should be any other worker
+    loop.run_until_complete(streamapp.run())

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "dam"
 version = "0.1.0"
-description = "Python Stream Processing toolkit"
+description = "Python Stream Processing minimalistic toolkit"
 authors = [
     "Marcos Schroh <schrohm@gmail.com>",
     "Santiago Fraire Willemoës <santiwilly@gmail.com>"
@@ -11,6 +11,7 @@ keywords = ["stream", "processing", "streaming", "async"]
 
 [tool.poetry.dependencies]
 python = "^3.7"
+aiokafka = "^0.5.2"
 
 [tool.poetry.dev-dependencies]
 pytest = "^5.3.5"