feat: example of integration with dataclasses-avroschema added

Author: marcosschroh

faust-project/Dockerfile

@@ -21,5 +21,3 @@ RUN pip3 install -e .
 ENTRYPOINT ["./wait_for_services.sh"]
 
 CMD ["./run.sh", "${WORKER}", "${WORKER_PORT}", "${CONFIG_CLASS}"]
-
-

faust-project/example/codecs/avro.py

@@ -3,6 +3,7 @@
 
 from simple_settings import settings
 
+from example.users.models import AdvanceUserModel
 
 # Initialize Schema Registry Client
 client = SchemaRegistryClient(url=settings.SCHEMA_REGISTRY_URL)
@@ -19,6 +20,14 @@
 
 avro_user_serializer = FaustSerializer(client, "users", avro_user_schema)
 
+# example of how to use it with dataclasses-avroschema
+avro_advance_user_serializer = FaustSerializer(
+    client, "advance_users", AdvanceUserModel.avro_schema())
+
 
 def avro_user_codec():
     return avro_user_serializer
+
+
+def avro_advance_user_codec():
+    return avro_advance_user_serializer

faust-project/example/users/agents.py

@@ -1,10 +1,11 @@
 import logging
 
 from example.app import app
-from example.codecs.avro import avro_user_serializer
-from example.users.models import UserModel
+from example.codecs.avro import avro_user_serializer, avro_advance_user_serializer
+from example.users.models import UserModel, AdvanceUserModel
 
 users_topic = app.topic('avro_users', partitions=1, value_type=UserModel)
+advance_users_topic = app.topic('advance_avro_users', partitions=1, value_type=AdvanceUserModel)
 
 logger = logging.getLogger(__name__)
 
@@ -20,4 +21,18 @@ async def users(users):
 async def publish_users():
     logger.info('PUBLISHING ON LEADER FOR USERS APP!')
     user = {"first_name": "foo", "last_name": "bar"}
-    await users.send(value=user, value_serializer=avro_user_serializer)
+    await users_topic.send(value=user, value_serializer=avro_user_serializer)
+
+
+@app.agent(advance_users_topic)
+async def advance_users(users):
+    async for user in users:
+        logger.info("Event received in topic avro_users")
+        logger.info(f"First Name: {user.first_name}, last name {user.last_name} age {user.age}")
+
+
+@app.timer(5.0, on_leader=True)
+async def advance_publish_users():
+    logger.info('PUBLISHING ON LEADER FOR USERS APP!')
+    user = {"first_name": "foo", "last_name": "bar", "age": 20}
+    await advance_users_topic.send(value=user, value_serializer=avro_advance_user_serializer)

faust-project/example/users/models.py

@@ -1,6 +1,14 @@
 import faust
 
+from dataclasses_avroschema import AvroModel
+
 
 class UserModel(faust.Record, serializer='avro_users'):
     first_name: str
     last_name: str
+
+
+class AdvanceUserModel(faust.Record, AvroModel, serializer='avro_advance_users'):
+    first_name: str
+    last_name: str
+    age: int

faust-project/setup.py

@@ -2,9 +2,12 @@
 
 requires = [
     "colorlog>=3.1.4",
-    "simple-settings",
+    "dataclasses-avroschema",
     "faust",
     "python-schema-registry-client",
+    "yarl<1.6.0,>=1.0",
+    "multidict<5.0,>=4.5",
+    "simple-settings",
     "typing-extensions",
 ]
 
@@ -34,6 +37,7 @@
         ],
         'faust.codecs': [
             'avro_users = example.codecs.avro:avro_user_codec',
+            'avro_advance_users = example.codecs.avro:avro_advance_user_codec',
         ],
     },
 )